Rumel created KAFKA-9547: ---------------------------- Summary: Kafka transaction - skip one offset when the application stops and be started again Key: KAFKA-9547 URL: https://issues.apache.org/jira/browse/KAFKA-9547 Project: Kafka Issue Type: Bug Components: clients Affects Versions: 2.4.0 Environment: I am using kafka-clients 2.4.0 and wurstmeister/kafka:2.12-2.3.0 Reporter: Rumel
To be fair, I have tested it with normal kafka without transaction scheme, and it does not skip the offset when I try to rerun the ProducerTest like a lot of times. {code:java} object ProducerTest extends LazyLogging { def main(args: Array[String]): Unit = { val props = new Properties() props.put("bootstrap.servers", "kafka.local:9092") props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer") props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer") props.put("acks", "all") props.put("retries", "3") val producer = new KafkaProducer[String, String](props) val record = new ProducerRecord[String, String]("zxc", "key", "value") val record2 = new ProducerRecord[String, String]("zxc", "key2", "value2") val record3 = new ProducerRecord[String, String]("zxc", "key3", "value3") producer.send(record) producer.send(record2) producer.send(record3) Thread.sleep(3000) } }{code} But when I enable transaction on producer, it will skip one offset when the ProducerTestWithTransaction application is rerun. Like when I first started it, it has an offset of 0,1,2 then after rerun, it will be 4,5,6 which skips 3, and so on and so forth. {code:java} object ProducerTestWithTransaction extends LazyLogging { def main(args: Array[String]): Unit = { val props = new Properties() props.put("bootstrap.servers", "kafka.local:9092") props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer") props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer") props.put("enable.idempotence", "true") props.put("transactional.id", "alona") props.put("acks", "all") props.put("retries", "3") val producer = new KafkaProducer[String, String](props) val record = new ProducerRecord[String, String]("wew", "key", "value") val record2 = new ProducerRecord[String, String]("wew", "key2", "value2") val record3 = new ProducerRecord[String, String]("wew", "key3", "value3") producer.initTransactions() try { producer.beginTransaction() producer.send(record) producer.send(record2) producer.send(record3) producer.commitTransaction() } catch { case e: ProducerFencedException => producer.close() case e: Exception => producer.abortTransaction(); } } }{code} Please enlighten me why this is happening? Is this the standard behavior when we are using transaction? Is there any workaround on this to not skip an offset. Thanks! -- This message was sent by Atlassian Jira (v8.3.4#803005)