reassign_partitions_test system test find idempotent message duplicated

[superhx]

AssertionError('Detected 9 duplicates even though idempotence was enabled.\n')

Consumer

sort console_consumer.stdout | uniq -c | sort -nk 1 | tail -10
      1 9999
      2 4570
      2 4571
      2 4572
      2 4573
      2 4574
      2 4575
      2 4576
      2 4577
      2 4578

Producer

{"timestamp":1699561616718,"name":"producer_send_success","key":null,"value":"4565","offset":114227,"topic":"test_topic","partition":1}
{"timestamp":1699561616723,"name":"producer_send_success","key":null,"value":"4566","offset":114228,"topic":"test_topic","partition":1}
{"timestamp":1699561616724,"name":"producer_send_success","key":null,"value":"4567","offset":114229,"topic":"test_topic","partition":1}
{"timestamp":1699561616726,"name":"producer_send_success","key":null,"value":"4568","offset":114230,"topic":"test_topic","partition":1}
date -d @1699561616
Fri Nov 10 04:26:56 CST 2023
{"timestamp":1699561616726,"name":"producer_send_success","key":null,"value":"4569","offset":114231,"topic":"test_topic","partition":1}
date -d @1699561619
Fri Nov 10 04:26:59 CST 2023
{"timestamp":1699561619024,"name":"producer_send_success","key":null,"value":"4570","offset":114241,"topic":"test_topic","partition":1}
{"timestamp":1699561619024,"name":"producer_send_success","key":null,"value":"4571","offset":114242,"topic":"test_topic","partition":1}
{"timestamp":1699561619024,"name":"producer_send_success","key":null,"value":"4572","offset":114243,"topic":"test_topic","partition":1}
{"timestamp":1699561619048,"name":"producer_send_success","key":null,"value":"4573","offset":114244,"topic":"test_topic","partition":1}
{"timestamp":1699561619048,"name":"producer_send_success","key":null,"value":"4574","offset":114245,"topic":"test_topic","partition":1}

grep -r 'osing partition' . | grep 'test_topic-1 ' |grep info
./ducker07/info/server.log:[2023-11-09 20:26:56,718] INFO [Partition test_topic-1 broker=4] Closing partition (kafka.cluster.Partition)
./ducker07/info/server.log:[2023-11-09 20:28:27,680] INFO [Partition test_topic-1 broker=4] Closing partition (kafka.cluster.Partition)
./ducker06/info/server.log:[2023-11-09 20:28:18,167] INFO [Partition test_topic-1 broker=3] Closing partition (kafka.cluster.Partition)
./ducker05/info/server.log:[2023-11-09 20:28:14,206] INFO [Partition test_topic-1 broker=2] Closing partition (kafka.cluster.Partition)

ducker07
KafkaService-1-140232534602608/ducker07/info
[2023-11-09 20:26:56,718] INFO [Partition test_topic-1 broker=4] Closing partition (kafka.cluster.Partition)
[2023-11-09 20:26:56,740] INFO [ElasticProducerStateManager partition=test_topic-1] Wrote producer snapshot at offset 114241 with 2 producer ids in 21 ms. (kafka.log.streamaspect.ElasticProducerStateManager)

ducker05
[2023-11-09 20:26:58,272] INFO [ElasticLog partition=test_topic-1 epoch=2] opened existing meta stream: stream_id=30 (kafka.log.streamaspect.ElasticLog$)
[2023-11-09 20:26:58,577] INFO [ElasticLog partition=test_topic-1 epoch=2] loaded partition meta: ElasticPartitionMeta{startOffset=113332, cleanerOffset=0, recoverOffset=114241, cleanedShutdown=true} (kafka.log.streamaspect.ElasticLog$)
[2023-11-09 20:26:58,578] INFO [ElasticLog partition=test_topic-1 epoch=2] loaded 2 producer snapshots, offsets(filenames) are Set(114241, 113332)  (kafka.log.streamaspect.ElasticLog$)
[2023-11-09 20:26:58,658] INFO [ElasticProducerStateManager partition=test_topic-1] Deleted producer snapshot file '113332' for partition test_topic-1 (kafka.log.streamaspect.ElasticProducerStateManager)
server.log-[2023-11-09 20:26:58,888] ERROR [ReplicaManager broker=2] Error processing append operation on partition test_topic-1 (kafka.server.ReplicaManager)
server.log:org.apache.kafka.common.errors.OutOfOrderSequenceException: Out of order sequence number for producer 1 at offset 114241 in partition test_topic-1: 900 (incoming seq. number), 908 (current end sequence number)

[superhx]

Reason: The Snapshot of ProducerStateManager only saves the last batchMetadata in ProducerStateEntry, causing the reloaded state to not recognize retry messages with a sequence smaller than the current broker sequence. When the client receives an OutOfOrderSequenceException, it bumps the epoch and attempts to resend, leading to duplicates.