Kafka使用者不使用事务性语义（isolation.level=read_commited)

Question

以下是我的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在其属性中有一个事务id，在推送一些消息之后，它将事务作为一个整体提交。以下是Kafka制作人的属性：-

log.info（“初始化属性”）；属性道具=新属性（）；

    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");

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;

我无法理解是否commit没有从生产者端正确地发生，从而导致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和纪元，在生产者发出的所有未来事务性消息中使用。