Kafka将KTable流到流INVALID_TOPIC_EXCEPTION

Question

我的流服务执行的操作很少：

stream
  .map(doSomething)
  .filter(filterSomething)
  .groupBy(groupMyStuffs)
  .aggregate(Map.empty[String, Object])(aggregation)
  .mapValues((k, v) => parseAggResults(k, v))
  .toStream
  .flatMap((_, v) => v)
  .to(outputTopic)

在进行测试时，我发现我的服务在mapvalues调用函数toStream后中断了，该函数将把我的数据写入由Kafka Streams将KTable转换为Kafka Streams创建的新主题。

我检查了KStreams创建的主题，主题就在那里：

myconsumergroup-KSTREAM-AGGREGATE-STATE-STORE-0000000005-changelog
myconsumergroup-KSTREAM-AGGREGATE-STATE-STORE-0000000005-repartition

我发现response有三个输入，即topicmetadata，我不知道第三个输入是什么：

(type=TopicMetadata, error=NONE, topic=myconsumergroup-KSTREAM-AGGREGATE-STATE-STORE-0000000005-repartition, isInternal=false, partitionMetadata=[(type=PartitionMetadata, error=NONE, partition=0, leader=xxx.xxx.xxx.xxx:9092 (id: 0 rack: null), replicas=xxx.xxx.xxx.xxx:9092 (id: 0 rack: null), isr=xxx.xxx.xxx.xxx:9092 (id: 0 rack: null), offlineReplicas=)])
(type=TopicMetadata, error=NONE, topic=myconsumergroup-KSTREAM-AGGREGATE-STATE-STORE-0000000005-changelog, isInternal=false, partitionMetadata=[(type=PartitionMetadata, error=NONE, partition=0, leader=xxx.xxx.xxx.xxx:9092 (id: 0 rack: null), replicas=xxx.xxx.xxx.xxx:9092 (id: 0 rack: null), isr=xxx.xxx.xxx.xxx:9092 (id: 0 rack: null), offlineReplicas=)])
(type=TopicMetadata, error=INVALID_TOPIC_EXCEPTION, topic=, isInternal=false, partitionMetadata=[])

为了确保所有内容都被覆盖，这里是我的配置：

logger.info(s"Loading Kafka configurations")
logger.info(s"Kafka Connection with: ${getEnvVar("KAFKA_PROTOBUF_CONN")}")
logger.info(s"Consumer Name: ${getEnvVar("CONSUMER_STREAM_NAME")}")
settings.put(StreamsConfig.APPLICATION_ID_CONFIG, getEnvVar("CONSUMER_STREAM_NAME"))
settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, getEnvVar("KAFKA_PROTOBUF_CONN"))
settings.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, getEnvVar("AUTO_OFFSET_RESET_CONFIG"))
settings.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass.getName)
settings.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.ByteArray().getClass.getName)
settings.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10.seconds)
settings.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, "0")
//Added to avoid messages created by old producers
settings.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, "org.apache.kafka.streams.processor.WallclockTimestampExtractor")

settings.put(StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG, "1000")
settings.put(StreamsConfig.producerPrefix(ProducerConfig.ACKS_CONFIG), "all")
settings.put(StreamsConfig.producerPrefix(ProducerConfig.MAX_REQUEST_SIZE_CONFIG), 20.mb)

if (!isLocalRun)
  settings.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, "3")

我的问题是,我们的部署正在工作,突然所有的东西都开始出现这个错误：

[kafka-producer-network-thread | myconsumergroup-1d0237ae-6caa-4cbd-aeaa-2154d2303b32-StreamThread-1-producer] WARN org.apache.kafka.clients.NetworkClient - [Producer clientId=myconsumergroup-1d0237ae-6caa-4cbd-aeaa-2154d2303b32-StreamThread-1-producer] Error while fetching metadata with correlation id 9 : {=INVALID_TOPIC_EXCEPTION}

Answer 1

大黄鸭又来袭了……

问题的答案是，输出主题的配置得到了一个空字符串。

因此.To(outputTopic)的生成器在MapValues执行完后就失败了