Gary Y. created KAFKA-6119: ------------------------------ Summary: Silent Data Loss in Kafka011 Transactional Producer Key: KAFKA-6119 URL: https://issues.apache.org/jira/browse/KAFKA-6119 Project: Kafka Issue Type: Bug Components: core, producer Affects Versions: 0.11.0.0, 0.11.0.1 Environment: openjdk version "1.8.0_144" OpenJDK Runtime Environment (Zulu 8.23.0.3-macosx) (build 1.8.0_144-b01) OpenJDK 64-Bit Server VM (Zulu 8.23.0.3-macosx) (build 25.144-b01, mixed mode) Reporter: Gary Y. Priority: Blocker
Kafka can lose data published by a transactional {{KafkaProducer}} under some circumstances, i.e., data that should be committed atomically may not be fully visible from a consumer with {{read_committed}} isolation level. *Steps to reproduce:* # Set {{transaction.timeout.ms}} to a low value such as {{100}} # Publish two messages in one transaction to different partitions of a topic with a sufficiently long time in-between the messages (e.g., 70 s). # Only the second message is visible with {{read_committed}} isolation level. See https://github.com/GJL/kafka011-transactional-producer-bug-demo/blob/master/src/main/java/com/garyyao/App.java for a full example. Detailed instructions can be found in the {{README.md}}: https://github.com/GJL/kafka011-transactional-producer-bug-demo *Why is this possible?* Because the transaction timeout is set to a low value, the transaction will be rolled back quickly after sending the first message. Indeed, in the broker the following logs could be found: {code} [2017-10-25 22:54:58,224] INFO [Transaction Coordinator 0]: Initialized transactionalId test-producer-1508964897483 with producerId 5 and producer epoch 0 on partition __transaction_state-10 (kafka.coordinator.transaction.TransactionCoordinator) [2017-10-25 22:55:24,260] INFO [Transaction Coordinator 0]: Completed rollback ongoing transaction of transactionalId: test-producer-1508964897483 due to timeout (kafka.coordinator.transaction.TransactionCoordinator) {code} After rollback the second message is sent to a different partition than the first message. Upon, transaction commit, {{org.apache.kafka.clients.producer.internals.TransactionManager}} may enqueue the request {{addPartitionsToTransactionHandler}}: {code} private TransactionalRequestResult beginCompletingTransaction(TransactionResult transactionResult) { if (!newPartitionsInTransaction.isEmpty()) enqueueRequest(addPartitionsToTransactionHandler()); EndTxnRequest.Builder builder = new EndTxnRequest.Builder(transactionalId, producerIdAndEpoch.producerId, producerIdAndEpoch.epoch, transactionResult); EndTxnHandler handler = new EndTxnHandler(builder); enqueueRequest(handler); return handler.result; } {code} As can be seen, the condition is fulfilled if {{newPartitionsInTransaction}} is non-empty. I suspect because the second message goes to a different partition, this condition is satisfied. In {{KafkaApis.scala}}, I can see that {{handleAddPartitionToTxnRequest}} eventually may call {{prepareAddPartitions}}: {code} def prepareAddPartitions(addedTopicPartitions: immutable.Set[TopicPartition], updateTimestamp: Long): TxnTransitMetadata = { val newTxnStartTimestamp = state match { case Empty | CompleteAbort | CompleteCommit => updateTimestamp case _ => txnStartTimestamp } prepareTransitionTo(Ongoing, producerId, producerEpoch, txnTimeoutMs, (topicPartitions ++ addedTopicPartitions).toSet, newTxnStartTimestamp, updateTimestamp) } {code} Note that the method's first argument {{newState}} of is always *Ongoing* here. I suspect that this puts the transaction, which should be aborted, to _Ongoing_ again. -- This message was sent by Atlassian JIRA (v6.4.14#64029)