仅一次两个Kafka集群

Question

我有 2 个Kafka集群。群集 A 和群集 B。这些集群是完全独立的。我有一个Spring启动应用程序，它侦听集群 A 上的主题，转换事件，然后将其生成到集群 B 上。我只需要一次，因为这些是金融事件。我注意到，对于我当前的应用程序，我有时会遇到重复的情况，也会错过一些事件。我试图尽我所能只实现一次。其中一篇帖子说，与Spring启动相比，flink将是一个更好的选择。我应该搬到闪光灯吗？请参阅下面的Spring代码。

消费者配置

@Configuration
public class KafkaConsumerConfig {

    @Value("${kafka.server.consumer}")
    String server;

    @Value("${kafka.kerberos.service.name:}")
    String kerberosServiceName;

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> config = new HashMap<>();

        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, server);
        config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, AvroDeserializer.class);
        config.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        
        // skip kerberos if no value is provided
        if (kerberosServiceName.length() > 0) {
          config.put("security.protocol", "SASL_PLAINTEXT");
          config.put("sasl.kerberos.service.name", kerberosServiceName);
        }

        return config;
    }

    @Bean
    public ConsumerFactory<String, AccrualSchema> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(),
                new AvroDeserializer<>(AccrualSchema.class));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, AccrualSchema> kafkaListenerContainerFactory(ConsumerErrorHandler errorHandler) {
        ConcurrentKafkaListenerContainerFactory<String, AccrualSchema> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setAutoStartup(true);
        factory.getContainerProperties().setAckMode(AckMode.RECORD);
        
        factory.setErrorHandler(errorHandler);
        
        return factory;
    }

    @Bean
    public KafkaConsumerAccrual receiver() {
        return new KafkaConsumerAccrual();
    }
}

生产者配置

    @Configuration
public class KafkaProducerConfig {

    @Value("${kafka.server.producer}")
    String server;

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, server);
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        config.put(ProducerConfig.ACKS_CONFIG, "all");
        config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "prod-1");
        config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");
        config.put(ProducerConfig.LINGER_MS_CONFIG, "10");
        config.put(ProducerConfig.BATCH_SIZE_CONFIG, Integer.toString(32*1024));
        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

}

Kafka生产者

    @Service
public class KafkaTopicProducer {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
        
    public void topicProducer(String payload, String topic) {
        kafkaTemplate.executeInTransaction(kt->kt.send(topic, payload));            
    }
}

Kafka消费者

public class KafkaConsumerAccrual {

    @Autowired
    KafkaTopicProducer kafkaTopicProducer;

    @Autowired
    Gson gson;
 
    @KafkaListener(topics = "topicname", groupId = "groupid", id = "listenerid")
    public void consume(AccrualSchema accrual,
            @Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partition, @Header(KafkaHeaders.OFFSET) Long offset,
            @Header(KafkaHeaders.CONSUMER) Consumer<?, ?> consumer) {       
        
        
        AccrualEntity accrualEntity = convertBusinessObjectToAccrual(accrual,partition,offset);

        kafkaTopicProducer.topicProducer(gson.toJson(accrualEntity, AccrualEntity.class), accrualTopic);

    }

    public AccrualEntity convertBusinessObjectToAccrual(AccrualSchema ao, Integer partition,
            Long offset) {
        //Transform code goes here
        return ae;
    }
}

Answer 1

根据KIP草案：https://cwiki.apache.org/confluence/display/KAFKA/KIP-656:MirrorMaker2精确一次语义

你可以期待的更好的事情是两个集群之间至少有一个语义。这意味着您必须在目标代理的消费者端防止重复。

例如，您可以确定一组属性，这些属性对于特定的时间窗口必须是唯一的。但我将真正依赖于您的用例。

Answer 2

确切地说，一旦跨集群不支持语义;关键是生成记录并在一个原子事务中提交消耗的偏移量。

这在您的环境中是不可能的，因为一个事务不能跨越两个集群。