如果无法访问接收器节点/主题，Kafka streams将关闭？

Question

我想测试使用处理器API的Kafka Streams从源代码读取和写入主题列表时的场景，一个或两个主题是不可访问的（失败测试：试图通过添加集群中不存在的1/2主题来模拟它）。

     topology.addSource("mysource","source_topic");
     topology.addProcessor("STREAM_PROCESSOR",()->new SourceProcessor(),"mysource");
     topology.addSink("SINK_1","TOPIC_1","STREAM_PROCESSOR");
     topology.addSink("SINK_2","TOPIC_2","STREAM_PROCESSOR");
     topology.addSink("SINK_3","TOPIC_3","STREAM_PROCESSOR"); // This topic is not present in cluster

      sourceContext.forward(eventName,eventMessage,To.child("sink1 or sink2 or sink3"));

我的理解是kafkaStreams应该为不存在的主题给出错误，并继续将记录转发到存在的主题1和主题2。

但我看到的行为是，它会产生以下错误：

     Exception in thread "StreamProcessor-56da56e4-4ab3-4ca3-bf48-b059558b689f-StreamThread-1" 
     org.apache.kafka.streams.errors.StreamsException: 
     task [0_0] Abort sending since an error caught with a previous record (timestamp 1592940025090) to topic "TOPIC_X" due to 
     org.apache.kafka.common.errors.TimeoutException: Topic "TOPIC_X" not present in metadata after 60000 ms.
     Timeout exception caught when sending record to topic "TOPIC_X". This might happen if the producer cannot send data to the
     Kafka cluster and thus, its internal buffer fills up. This can also happen if the broker is slow to respond, if the network connection to the
     broker was interrupted, or if similar circumstances arise. You can increase producer parameter `max.block.ms` to increase this timeout.

这是模拟不可达主题或主题不存在问题的正确方式吗？为什么即使在处理Streams和拓扑异常时，Kafka流也会因上述错误而关闭。如果某个接收器主题因某种原因不可用或不可访问，kafka流不应该关闭，对吗？。好心建议

关于上面的错误，我想在捕获StreamsException到错误主题时转发错误，但是kafkastreams会过早停止。

catch(StreamsException e)
{
    context.forward("","",Error_Topic)
}

这是预期的行为吗？

参考：https://docs.confluent.io/current/streams/developer-guide/manage-topics.html#user-主题这是否意味着kafkastreams拓扑中不允许不存在的主题作为接收节点。请确认。

Answer 1

根据设计，如果Kafka流不能写入水槽主题，它就会关闭。原因是默认情况下，Kafka Streams保证至少一次处理语义学，如果它不能将数据写入一个接收器主题，但将继续，至少一次处理将被违反，因为接收器主题中会有数据丢失。

有一个产品。例外处理程序可能有帮助的配置。它允许您在将数据写入输出主题时吞下某些异常。但是，请注意，这意味着相应主题的数据丢失。