error-handling - 如何处理未授权访问Kafka Streams中的主题…

Question

情况如下。

我们在Kafka Broker中设置了SSL + ACL。

我们正在建立流，该流从两个主题读取消息:

KStream<String, String> stringInput 
    = kBuilder.stream( STRING_SERDE, STRING_SERDE, inTopicName );

stringInput
    .filter( streamFilter::passOrFilterMessages )
    .map( processor )
    .to( outTopicName );

它完成了两次(在循环中)。
然后我们设置常规错误处理程序:

streams.setUncaughtExceptionHandler( ( Thread t, Throwable e ) -> {
                    synchronized ( this ) {
                        LOG.fatal( ... );
                        this.stop();
                    }
                }
        );

问题如下。例如，如果在一个主题中证书不再有效。流抛出异常未经授权访问主题...
到现在为止还挺好。

但是异常是由通用错误处理程序处理的，因此即使第二个主题没有问题，整个应用程序也会停止运行。

问题是，如何按主题处理此异常？
如何避免由于 一个 单个主题存在授权问题而导致整个应用程序有时停止的情况？

我了解，如果经纪人不可用，则完整的应用程序可能会停止。但是，如果只有一个主题不可用，那么单个流将停止，并且无法完成应用程序，或者？

Answer 1

按照设计，Kafka Streams将拓 flutter 视为一个，并且无法区分这两个部分。对于您的特定情况，当您循环并构建到独立的管道时，可以并行运行两个KafkaStreams实例(在同一应用程序/JVM中)以将彼此隔离。因此，如果一个失败，则另一个不受影响。您将需要为两个实例使用两个不同的application.id。