属性名是interceptor.classes，而不是interceptors.classes

这个问题已经很老了，所以我想你同时也找到了解决办法。然而，为了以防万一，我发现我的 ProducerInterceptor 类，它根据消息的内容将消息分派到不同的主题，除非我的流已经有了指定的输出，否则不会被调用。
我的第一次尝试看起来像这样，因为我认为我不需要指定输出主题。这不起作用：

val builder: KStreamBuilder = new KStreamBuilder
val input = builder.stream("input-topic")

val stream: KafkaStreams = new KafkaStreams(builder, streamsConfigWithProducerInterceptor)
stream.start()

但事实上：

val builder: KStreamBuilder = new KStreamBuilder
val input = builder.stream("input-topic").through("dummy-output-topic")

val stream: KafkaStreams = new KafkaStreams(builder, streamsConfigWithProducerInterceptor)
stream.start()

值得一提的是，没有什么可以发表的 dummy-output-topic 在第二个例子中，使用 to 而不是 through 似乎也是这样。
就我而言，我是在 map 在使用拦截器将记录分派到不同的主题之前更改记录，因此我的代码实际上更像这样：

val builder: KStreamBuilder = new KStreamBuilder
val input = builder.stream("input-topic")
    .map(new CustomKeyValueMapper)
    .through("dummy-output-topic")

val stream: KafkaStreams = new KafkaStreams(builder, streamsConfigWithProducerInterceptor)
stream.start()

我希望这些例子能帮助任何与我一起工作的人 ProducerInterceptor 他犯了和我一样的错误。