Kafka Streams KTable to Stream无效的\u TOPIC_异常

我有一个Kafka流服务,显然没有任何原因开始有一个特定的错误,甚至在本地运行。

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流创建的一个新主题,该主题将把KTable转换为Kafka流。

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

然后我试着在这个问题上再深入一点。我开始调试Kafka客户机类,以确定出现此错误的原因,我发现在以下代码行:

https://github.com/apache/kafka/blob/2.0/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java#L966

我发现 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}

我想找到的是什么,或者说,这不是什么问题。