Kafka ConsumerInterceptor 所需的配置

Question

我是卡夫卡的新手。我已经建立了我的环境来生产和消费记录。但是，我的目标是在将记录发送给目标消费者之前拦截记录并修改它们的值。 现在，为了确保环境设置良好以拦截记录，我正在编写一个简单的 ConsumerInterceptor，它将拦截记录并打印它们的值。 我的 configure() 方法应该实现什么来启用我的 consumerInterceptor？我应该添加/修改哪些其他配置以及在哪里？

public class SimpleConsumerInterceptors<K, V> implements ConsumerInterceptor<K, V>{
    private String clientId;

    public void configure(final Map<String, ?> configs) {

// What configurations required to enable my consumerInterceptor?
        
    }

    public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records) {
        ConsumerRecords<K, V> interceptRecords = records;
        for (TopicPartition partition : records.partitions()) {
            String topic = partition.topic();
            List<ConsumerRecord<K, V>> recordsInPartition = records.records(partition);
            for (ConsumerRecord<K, V> record : recordsInPartition) {
                System.out.println("onConsume:");
                System.out.println(record.value());
            }
        }
                 
        return interceptRecords;
    }

    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        System.out.println("onCommit")
    }

    @Override
    public void close() {
        System.out.println("close")
        this.close();
    }
}

Answer 1

您所要做的就是将类名添加到interceptor.classes consumer property、ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG。

如果想在Kafka创建实例后配置拦截器，只需要实现configure()即可。

传递给方法的映射是消费者属性的映射。

所以，假设您想让日志记录成为可选；将my.interceptor.logging.enabled=true添加到消费者配置中，并在configure()方法中使用它来配置是否记录记录。