Kafka Consumer 不使用事务语义（isolation.level = read_commited）

Question

当 Kafka 消费者在属性中具有事务语义时，它无法使用任何消息。但是当我删除该属性或将该属性更新为 read_uncommited 时，它会消耗消息。

以下是我的 Kafka 消费者属性：-

Properties props = new Properties();
    props.put("bootstrap.servers", "10.2.200.15:9092");
    String consumeGroup = "cg3";
    props.put("group.id", consumeGroup);
    // Below is a key setting to turn off the auto commit.
    props.put("enable.auto.commit", "false");
    props.put("heartbeat.interval.ms", "2000");
    props.put("session.timeout.ms", "6001");
    // Control maximum data on each poll, make sure this value is bigger than the
    // maximum // single message size
    props.put("max.partition.fetch.bytes", "140");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("isolation.level","read_committed");

Kafka Producer 在其属性中有一个 Transactional id，并且在推送一些消息后，它正在提交整个事务。以下是 Kafka 生产者属性：-

log.info("初始化属性"); 属性 props = new Properties();

    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, System.getenv(KafkaConstants.KAFKA_URL));
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
            "org.apache.kafka.common.serialization.StringSerializer");
    props.put("linger.ms", 1000);
    props.put("acks", "all");
    // props.put("request.timeout.ms",30000);
    props.put("retries", 3);
    props.put("retry.backoff.ms", 1000);
    props.put("max.in.flight.requests.per.connection", 1); // if its greater than 1, it can change the order or records. Maximum no. of unacknowledge request a client can send.
    props.put("enable.idempotence", true);
    props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"Transaction8");

sn-p下面负责提交事务：-

public boolean send(ProducerRecordImpl record) {
    try {
        producer.beginTransaction();
        for (int i = 0; i < 10; i++) {


            Future<RecordMetadata> futureResult = producer
                    .send(new ProducerRecord<String, String>(record.getTopic(), record.getPayload()));
            /*
             * It will wait till the thread execution completes and return true.
             */
            //RecordMetadata ack = futureResult.get();
            //log.debug("RecordMetadta offset {} and partiton {} ", ack.offset(), ack.partition());
        }
        producer.commitTransaction();
        log.info("Commited");


        return true;

我无法理解提交是否没有从生产者端正确发生，这导致 Kafka 消费者无法通过事务语义读取它，或者 Kafka 消费者端仍然存在问题。

任何帮助将不胜感激。

Answer 1

你需要先调用 producer.initTransactions()。否则，您的生产者不会发布事务性消息。

来自https://kafka.apache.org/0110/javadoc/org/apache/kafka/clients/producer/KafkaProducer.html#initTransactions()

transactional.id 时需要在任何其他方法之前调用 在配置中设置。此方法执行以下操作： 1。 确保由以前的实例发起的任何事务 具有相同 transactional.id 的生产者已完成。如果上一个 实例因正在进行的事务而失败，它将是 中止。如果最后一个事务已开始完成，但尚未完成 完成，此方法等待其完成。 2.获取内部 生产者 id 和 epoch，用于所有未来的事务消息 由生产者发行。

Answer 2

我在 Kafka 上测试 Transactions 时遇到了同样的问题。问题是操作系统。我使用 Windows 10 运行 Kafka 代理，当将代理配置为“read_committed”时，消费者看不到任何已提交的事务，一旦我将代理移至 Linux，事务（和消费者）就开始工作。 顺便说一句，Kafka 没有在日志中显示任何错误。 希望对您有所帮助。