Spring-Kafka多主题差异化配置：方案对比及优化咨询

我们的API通过Spring-Kafka连接多个Kafka主题，负责消息的生产与消费。除了能在运行时指定主题名称外，每个主题还有一个差异化配置项——消息序列化/反序列化的加密密钥，这导致无法为所有主题复用同一套生产者/消费者配置。

方案一：每个主题对应独立KafkaTemplate Bean #

实现代码 #

@Configuration
@EnableKafka
@RequiredArgsConstructor
public class KafkaProducerConfig {

    private final KafkaProperties properties;

    @Bean(name = TOPIC_1_PRODUCER)
    public KafkaTemplate<String, GenericRecord> kafkaTemplateTopic1() {
        Map<String, Object> producerProps = properties.buildProducerProperties();
        producerProps.put(ENCRYPTION_KEY, "encryption_key_for_topic_1");
        return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(producerProps));
    }

    @Bean(name = TOPIC_2_PRODUCER)
    public KafkaTemplate<String, GenericRecord> kafkaTemplateTopic2() {
        Map<String, Object> producerProps = properties.buildProducerProperties();
        producerProps.put(ENCRYPTION_KEY, "encryption_key_for_topic_2");
        return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(producerProps));
    }

    @Bean(name = TOPIC_3_PRODUCER)
    public KafkaTemplate<String, GenericRecord> kafkaTemplateTopic3() {
        Map<String, Object> producerProps = properties.buildProducerProperties();
        producerProps.put(ENCRYPTION_KEY, "encryption_key_for_topic_3");
        return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(producerProps));
    }
}

方案说明 #

为每个需要生产消息的主题创建独立的KafkaTemplate Bean，每个主题对应一个实例，注入到对应主题的消息生产服务中使用。

方案二：运行时动态生成KafkaTemplate实例 #

配置类代码 #

@Configuration
@EnableKafka
@RequiredArgsConstructor
public class KafkaProducerConfig {

    private final KafkaProperties properties;

    @Bean
    public ProducerFactory<String, GenericRecord> producerFactory() {
        Map<String, Object> producerProps = properties.buildProducerProperties();
        return new DefaultKafkaProducerFactory<>(producerProps);
    }
    
}

发布服务类代码 #

@Slf4j
@Service
@RequiredArgsConstructor
public class KafkaProducer {

    private final ProducerFactory<String, GenericRecord> producerFactory;
    private final TopicResolver topicResolver;

    public void sendMessage(GenericRecord genericRecord) {
        TopicInfo topicInfo = topicResolver.resolve(genericRecord);

        Map<String, Object> configOverrides = new HashMap<>();
        configOverrides.put(ENCRYPTION_KEY, "encryption_key_for_resolved_topic");
        KafkaTemplate<String, GenericRecord> kafkaTemplate = new KafkaTemplate<>(producerFactory, configOverrides);

        ProducerRecord<String, GenericRecord> producerRecord = new ProducerRecord<>(topicInfo.getTopicName(), genericRecord);
        ListenableFuture<SendResult<String, GenericRecord>> listenableFuture = kafkaTemplate.send(producerRecord);
        listenableFuture.addCallback(new KafkaSendCallback<String, GenericRecord>() {
            @Override
            public void onSuccess(SendResult<String, GenericRecord> result) {
                log.info("Kafka message successfully sent");
            }

            @Override
            public void onFailure(@NonNull KafkaProducerException ex) {
                log.error("Kafka message failed to be sent with error {}", ex.getMessage());
            }
        });
    }
}

方案说明 #

仅维护一个ProducerFactory Bean，在发送每条消息时，根据解析出的主题动态生成带有对应加密密钥的KafkaTemplate实例。但这种每条消息创建新实例的方式，我们不确定是否符合Spring-Kafka的设计意图，以及会对Kafka性能造成何种影响。

咨询问题 #

1. 上述两种方案各有何优劣？
2. 是否存在其他方案，能够在使用单个KafkaTemplate Bean/单个Kafka生产者的同时，实现运行时传递差异化配置（如加密密钥）？
3. 消费者端也存在相同的差异化加密密钥需求，该如何处理？

内容的提问来源于stack exchange，提问作者Diana Fagateanu