即使未调用producer commit in，Kafka事务生产者也会写入消息

Question

对我来说，Kafka事务制作人的行为就像一个普通的制作人，每次消息调用send时，在主题上都可以看到Meesage。也许我错过了一些基本的东西。我希望只有在调用producer commit方法之后，消息才会出现在主题中。在我下面的代码中。commitTransactions（）已被注释掉，但我仍能收到主题中的消息。谢谢你的指点。

  public static void main(String[] args) {
        try {
            Properties producerConfig = new Properties();
            producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "...");
            producerConfig.put(ProducerConfig.CLIENT_ID_CONFIG, "transactional-producer-1");
            producerConfig.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // enable idempotence
            producerConfig.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "test-transactional-id-1"); // set transaction id
            producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
            producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

            Producer<String, String> producer = new KafkaProducer<>(producerConfig);

            producer.initTransactions(); //initiate transactions
            try {
                producer.beginTransaction(); //begin transactions
                for (Integer i = 0; i < 1000; i++) {
                    producer.send(new ProducerRecord<String, String>("t_test", i.toString(), "value_" + i));

                }
                // producer.commitTransaction(); //commit

            } catch (KafkaException e) {
                // For all other exceptions, just abort the transaction and try again.
                producer.abortTransaction();
            }

            producer.close();
        } catch (Exception e) {
            System.out.println(e.toString());
        }
    }

Answer 1

当涉及到交易在Kafka你需要考虑生产者/消费者对。正如您所观察到的，生产者本身只是产生数据，要么提交事务，要么不提交事务。

只有在与消费者交互时，您才能通过将Kafka消费者配置isolation.level设置为read_committed（默认情况下设置为read_uncommitted）来完成事务。此配置描述为：

隔离级别：控制如何读取以事务方式写入的消息。如果设置为read_committed，则消费者。poll（）将仅返回已提交的事务性消息。如果设置为read_uncommitted（默认值），则消费者。poll（）将返回所有消息，甚至是已中止的事务性消息。非事务性消息将在任一模式下无条件返回。