[jira] [Commented] (KAFKA-19880) First batch in producer epoch can be appended out-of-order even for idempotent producer

["Travis Bischel (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-19880?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18064344#comment-18064344 ] 

Travis Bischel commented on KAFKA-19880:
----------------------------------------

Likely related: KAFKA-14312

> First batch in producer epoch can be appended out-of-order even for idempotent producer
> ---------------------------------------------------------------------------------------
>
>                 Key: KAFKA-19880
>                 URL: https://issues.apache.org/jira/browse/KAFKA-19880
>             Project: Kafka
>          Issue Type: Bug
>          Components: core, producer 
>    Affects Versions: 4.0.1, 4.1.0
>            Reporter: Ivan Yurchenko
>            Priority: Minor
>
> I observe the following behavior. A Java producer with enable.idempotence=true and max.in.flight.requests.per.connection=5 sends a number of records, enough to form several batches in several Produce requests. The topic is empty, freshly created.
> {code:java}
> [2025-11-11 11:41:19,125] DEBUG [Producer clientId=producer-1] Assigned producerId 0 and producerEpoch 0 to batch with base sequence 0 being sent to partition input-topic-0 (org.apache.kafka.clients.producer.internals.RecordAccumulator)
> [2025-11-11 11:41:19,132] DEBUG [Producer clientId=producer-1] Sending PRODUCE request with header RequestHeader(apiKey=PRODUCE, apiVersion=13, clientId=producer-1, correlationId=21, headerVersion=2) and timeout 30000 to node 3: {acks=-1,timeout=30000,partitionSizes=[4rrMdxpZQnyDKO4QKjPq-A:input-topic-0=16383]} (org.apache.kafka.clients.NetworkClient)
> [2025-11-11 11:41:19,133] DEBUG [Producer clientId=producer-1] Assigned producerId 0 and producerEpoch 0 to batch with base sequence 1458 being sent to partition input-topic-0 (org.apache.kafka.clients.producer.internals.RecordAccumulator)
> [2025-11-11 11:41:19,133] DEBUG [Producer clientId=producer-1] Sending PRODUCE request with header RequestHeader(apiKey=PRODUCE, apiVersion=13, clientId=producer-1, correlationId=22, headerVersion=2) and timeout 30000 to node 3: {acks=-1,timeout=30000,partitionSizes=[4rrMdxpZQnyDKO4QKjPq-A:input-topic-0=16377]} (org.apache.kafka.clients.NetworkClient)
> [2025-11-11 11:41:19,133] DEBUG [Producer clientId=producer-1] Assigned producerId 0 and producerEpoch 0 to batch with base sequence 2823 being sent to partition input-topic-0 (org.apache.kafka.clients.producer.internals.RecordAccumulator)
> [2025-11-11 11:41:19,133] DEBUG [Producer clientId=producer-1] Sending PRODUCE request with header RequestHeader(apiKey=PRODUCE, apiVersion=13, clientId=producer-1, correlationId=23, headerVersion=2) and timeout 30000 to node 3: {acks=-1,timeout=30000,partitionSizes=[4rrMdxpZQnyDKO4QKjPq-A:input-topic-0=16377]} (org.apache.kafka.clients.NetworkClient)
> [2025-11-11 11:41:19,133] DEBUG [Producer clientId=producer-1] Assigned producerId 0 and producerEpoch 0 to batch with base sequence 4188 being sent to partition input-topic-0 (org.apache.kafka.clients.producer.internals.RecordAccumulator)
> [2025-11-11 11:41:19,135] DEBUG [Producer clientId=producer-1] Sending PRODUCE request with header RequestHeader(apiKey=PRODUCE, apiVersion=13, clientId=producer-1, correlationId=24, headerVersion=2) and timeout 30000 to node 3: {acks=-1,timeout=30000,partitionSizes=[4rrMdxpZQnyDKO4QKjPq-A:input-topic-0=16377]} (org.apache.kafka.clients.NetworkClient)
> [2025-11-11 11:41:19,135] DEBUG [Producer clientId=producer-1] Assigned producerId 0 and producerEpoch 0 to batch with base sequence 5553 being sent to partition input-topic-0 (org.apache.kafka.clients.producer.internals.RecordAccumulator)
> [2025-11-11 11:41:19,135] DEBUG [Producer clientId=producer-1] Sending PRODUCE request with header RequestHeader(apiKey=PRODUCE, apiVersion=13, clientId=producer-1, correlationId=25, headerVersion=2) and timeout 30000 to node 3: {acks=-1,timeout=30000,partitionSizes=[4rrMdxpZQnyDKO4QKjPq-A:input-topic-0=16377]} (org.apache.kafka.clients.NetworkClient){code}
>  
> The cluster and the target broker are experiencing some connectivity issue affecting metadata propagation. The target broker doesn't think it's the leader of the partition for some time, so it rejects first requests:
>  
> {code:java}
> [2025-11-11 11:41:20,456] DEBUG [KafkaApi-3] Produce request with correlation id 21 from client producer-1 on partition 4rrMdxpZQnyDKO4QKjPq-A:input-topic-0 failed due to org.apache.kafka.common.errors.NotLeaderOrFollowerException (kafka.server.KafkaApis)
> [2025-11-11 11:41:24,876] DEBUG [KafkaApi-3] Produce request with correlation id 22 from client producer-1 on partition 4rrMdxpZQnyDKO4QKjPq-A:input-topic-0 failed due to org.apache.kafka.common.errors.NotLeaderOrFollowerException (kafka.server.KafkaApis)
> {code}
> Then it finds out it's the leader of the partition, but doesn't yet learn about the other replica:
>  
> {code:java}
> [2025-11-11 11:41:45,498] ERROR [ReplicaManager broker=3] Error processing append operation on partition 4rrMdxpZQnyDKO4QKjPq-A:input-topic-0 (kafka.server.ReplicaManager)
> org.apache.kafka.common.errors.NotEnoughReplicasException: The size of the current ISR : 1 is insufficient to satisfy the min.isr requirement of 2 for partition input-topic-0, live replica(s) broker.id are : Set(3)
> [2025-11-11 11:41:45,498] DEBUG [KafkaApi-3] Produce request with correlation id 23 from client producer-1 on partition 4rrMdxpZQnyDKO4QKjPq-A:input-topic-0 failed due to org.apache.kafka.common.errors.NotEnoughReplicasException (kafka.server.KafkaApis){code}
> And finally when it comes to the request with correlation ID 24, it succeeds. So the first written batch is logically out-of-sequence, but accepted.
>  
> The previously failed batches may be retried later, with new producer epoch (like happened in my case), but this doesn't matter: the intended order of records is violated.
>  
> I'm able to live reproduce this quite reliably under [Antithesis|https://antithesis.com/]. Probably, it won't be easy to make a local setup to catch this with good frequency. But I boiled this down to the following unit test (in ReplicaManagerTest):
> {noformat}
>   @Test
>   def testX(): Unit = {
>     val mockLogMgr = TestUtils.createLogManager(config.logDirs.asScala.map(new File(_)))
>     val node = new Node(0, "host0", 0)
>     mockGetAliveBrokerFunctions(metadataCache, Seq(node))
>     val rm = new ReplicaManager(
>       metrics = metrics,
>       config = config,
>       time = time,
>       scheduler = new MockScheduler(time),
>       logManager = mockLogMgr,
>       quotaManagers = quotaManager,
>       metadataCache = new KRaftMetadataCache(config.brokerId, () => KRaftVersion.KRAFT_VERSION_0),
>       logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size),
>       alterPartitionManager = alterPartitionManager,
>       threadNamePrefix = Option(this.getClass.getName))
>     try {
>       val tp = new TopicPartition(topic, 0)
>       val partition = rm.createPartition(tp)
>       partition.createLogIfNotExists(isNew = false, isFutureReplica = false,
>         new LazyOffsetCheckpoints(rm.highWatermarkCheckpoints.asJava), None)
>       rm.becomeLeaderOrFollower(0, new LeaderAndIsrRequest.Builder(0, 0, brokerEpoch,
>         Seq(new LeaderAndIsrRequest.PartitionState()
>           .setTopicName(topic)
>           .setPartitionIndex(0)
>           .setControllerEpoch(0)
>           .setLeader(config.brokerId())
>           .setLeaderEpoch(0)
>           .setIsr(Seq[Integer](config.brokerId()).asJava)
>           .setPartitionEpoch(0)
>           .setReplicas(Seq[Integer](config.brokerId()).asJava)
>           .setIsNew(false)).asJava,
>         Collections.singletonMap(topic, topicId),
>         Set(node).asJava).build(), (_, _) => ())
>       rm.getPartitionOrException(tp)
>         .localLogOrException
>       val topicIdPartition = new TopicIdPartition(Uuid.ZERO_UUID, partition.topicPartition)
>       val producerId = 10L
>       val producerEpoch: Short = 10
>       val r0 = MemoryRecords.withIdempotentRecords(Compression.NONE, producerId, producerEpoch, 0, new SimpleRecord("record 0".getBytes()))
>       val r1 = MemoryRecords.withIdempotentRecords(Compression.NONE, producerId, producerEpoch, 1, new SimpleRecord("record 1".getBytes()))
> //      val order = List(r0, r1)  // normal order -- no ordering violation
>       val order = List(r1, r0)  // r1 will be accepted out of order; r0 will be retried later with newer epoch and accepted
>       def callback(responseStatus: Map[TopicIdPartition, PartitionResponse]): Unit = {
>         println(responseStatus(topicIdPartition).error)
>       }
>       for (r <- order) {
>         rm.appendRecords(
>           timeout = 1000,
>           requiredAcks = -1,
>           internalTopicsAllowed = false,
>           origin = AppendOrigin.CLIENT,
>           entriesPerPartition = Map(topicIdPartition -> r),
>           responseCallback = callback,
>         )
>       }
>     } finally {
>       rm.shutdown(checkpointHW = false)
>     }
>   }
> {noformat}
>  
> As far as I understand, it isn't necessary for the partition to be empty. It's enough for the producer state to be empty and the first requests in the producer epoch to fail like above.
> If a first request in the producer epoch succeeds, the situation will not be possible. For example, we send r1, r2, r3. r1 succeeds, r2 fails with NotLeaderOrFollowerException, r3 won't be accepted and will fail with OutOfOrderSequenceException.
> I attribute this difference to [this validation on the broker side|https://github.com/apache/kafka/blob/3479ce793bafc6a1c42e6afa77e3fbfc3a36c80c/storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerAppendInfo.java#L144-L148].
>  
> AFAIU, this is a bug, at least according to how I read the documentation. I would expect enable.idempotence to prevent this from happening. Am I right or it is acceptable and we should improve the documentation instead?
> Can we require on the broker side that the first batch in a producer epoch always starts with the sequence 0?



--
This message was sent by Atlassian Jira
(v8.20.10#820010)