相当于Spring Cloud Stream Kafka Streams Binder的ChannelInterceptor

Question

我们在Spring Boot的基础上开发了一个内部公司框架，我们希望通过Spring Cloud Stream支持Kafka Streams。我们需要自动向所有出站消息注入一些头。我们通过标准的Spring Cloud Stream Kafka Binder注册了一个定制的通道拦截器，实现了这一点，但这不适用于Kafka Streams，因为它们似乎遵循不同的路径。

对于Spring Cloud Stream Kafka Streams binder，是否有任何等效于Channel Interceptor？

我找到了这个定制器/配置器：

java prettyprint-override"> @Bean
  public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() {
    return factoryBean -> {

      factoryBean.setInfrastructureCustomizer(new KafkaStreamsInfrastructureCustomizer() {
        @Override
        public void configureBuilder(final StreamsBuilder builder) {
          
        }

        @Override
        public void configureTopology(final Topology topology) {
          
        }
      });

      factoryBean.setKafkaStreamsCustomizer(new KafkaStreamsCustomizer() {
        @Override
        public void customize(final KafkaStreams kafkaStreams) {
          
        }
      });

    };
  }

我的最后一个想法是使用配置拓扑方法自动修改拓扑，并在最后一个接收节点之前插入一个转换器，但为了添加这个新节点，我必须指定父节点，所以我需要知道最后一个接收节点的名称，也许所有节点的名称都是由Kafka Streams自动生成的。。。唯一的方法是使用拓扑。descripe（）方法，并可能解析字符串输出。。。与简单的通道拦截器相比，这听起来太复杂了。

有什么想法吗？




Answer 1

您可以将一个producerInterceptor添加到流配置中。

/**
 * A plugin interface that allows you to intercept (and possibly mutate) the records received by the producer before
 * they are published to the Kafka cluster.
 * <p>
 * This class will get producer config properties via <code>configure()</code> method, including clientId assigned
 * by KafkaProducer if not specified in the producer config. The interceptor implementation needs to be aware that it will be
 * sharing producer config namespace with other interceptors and serializers, and ensure that there are no conflicts.
 * <p>
 * Exceptions thrown by ProducerInterceptor methods will be caught, logged, but not propagated further. As a result, if
 * the user configures the interceptor with the wrong key and value type parameters, the producer will not throw an exception,
 * just log the errors.
 * <p>
 * ProducerInterceptor callbacks may be called from multiple threads. Interceptor implementation must ensure thread-safety, if needed.
 * <p>
 * Implement {@link org.apache.kafka.common.ClusterResourceListener} to receive cluster metadata once it's available. Please see the class documentation for ClusterResourceListener for more information.
 */
public interface ProducerInterceptor<K, V> extends Configurable {

您可以在那里修改记录的标题。

（您也可以将此技术用于消息通道绑定器，而不是通道拦截器）。