KafkaEOS读取-处理-写入丢失了一些消息

使用Kafka事务来确保EOS语义。在读取记录之前，需要设置producer.beginTransaction()来启动事务，并在写入完所有记录后，使用producer.commitTransaction()提交事务。以下是使用Kafka事务确保EOS的代码示例：

// 创建Kafka生产者
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("transactional.id", "my-transactional-id");
Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());

// 启动事务
producer.initTransactions();
try {
    producer.beginTransaction();
    
    // 从Kafka读取记录
    Consumer<String, String> consumer = new KafkaConsumer<>(props, new StringDeserializer(), new StringDeserializer());
    consumer.subscribe(Arrays.asList("my-topic"));
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            // 处理记录
            
            // 将记录写回到Kafka中
            ProducerRecord<String, String> outputRecord = new ProducerRecord<>("output-topic", record.key(), record.value());
            producer.send(outputRecord);
        }
    }
    
    // 提交事务
    producer.commitTransaction();
} catch (Exception e) {
    // 回滚事务
    producer.abortTransaction();
    e.printStackTrace();
} finally {
    producer.close();
}