[INFO] Scanning for projects... [INFO] [INFO] ----------< com.ankurm.kafka:spring-kafka-exactly-once-boot4 >---------- [INFO] Building Exactly-once with Spring Kafka on Spring Boot 4 1.0.0 [INFO] from pom.xml [INFO] --------------------------------[ jar ]--------------------------------- [INFO] [INFO] --- resources:3.5.0:resources (default-resources) @ spring-kafka-exactly-once-boot4 --- [INFO] Copying 1 resource from src\main\resources to target\classes [INFO] Copying 0 resource from src\main\resources to target\classes [INFO] [INFO] --- compiler:3.15.0:compile (default-compile) @ spring-kafka-exactly-once-boot4 --- [INFO] Nothing to compile - all classes are up to date. [INFO] [INFO] --- resources:3.5.0:testResources (default-testResources) @ spring-kafka-exactly-once-boot4 --- [INFO] Copying 1 resource from src\test\resources to target\test-classes [INFO] [INFO] --- compiler:3.15.0:testCompile (default-testCompile) @ spring-kafka-exactly-once-boot4 --- [INFO] Recompiling the module because of changed source code. [INFO] Compiling 8 source files with javac [debug parameters release 25] to target\test-classes [INFO] [INFO] --- surefire:3.5.6:test (default-test) @ spring-kafka-exactly-once-boot4 --- [INFO] Using auto detected provider org.apache.maven.surefire.junitplatform.JUnitPlatformProvider [INFO] [INFO] ------------------------------------------------------- [INFO] T E S T S [INFO] ------------------------------------------------------- [INFO] Running com.ankurm.kafka.eos.SmokeTest Bootstrap metadata: BootstrapMetadata(records=[ApiMessageAndVersion(FeatureLevelRecord(name='metadata.version', featureLevel=30) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='eligible.leader.replicas.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='group.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='share.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='streams.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='transaction.version', featureLevel=2) at version 0)], metadataVersionLevel=30, source=format command) Formatting metadata directory C:\Users\Ankur\AppData\Local\Temp\kafka-15096196392826061749\combined_0_0 with metadata.version 4.3-IV0. 22:22:09.287 WARN o.a.k.c.QuorumController - [QuorumController id=0] Performing controller activation. The metadata log appears to be empty. Appending 6 bootstrap record(s) in metadata transaction at metadata.version 4.3-IV0 from bootstrap source 'testkit'. ======================================================================== Environment ======================================================================== java.version : 25.0.3 kafka-clients : 4.2.1 embedded broker : EmbeddedKafkaKraftBroker bootstrap servers : localhost:63356 cluster id : JmSQ3q0IRuilOAxcUgBP6w nodes : 1 topics : [smoke] [INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 2.729 s -- in com.ankurm.kafka.eos.SmokeTest [INFO] Running com.ankurm.kafka.eos._01_idempotence.IdempotentProducerTest Bootstrap metadata: BootstrapMetadata(records=[ApiMessageAndVersion(FeatureLevelRecord(name='metadata.version', featureLevel=30) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='eligible.leader.replicas.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='group.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='share.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='streams.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='transaction.version', featureLevel=2) at version 0)], metadataVersionLevel=30, source=format command) Formatting metadata directory C:\Users\Ankur\AppData\Local\Temp\kafka-9366343747489431922\combined_0_0 with metadata.version 4.3-IV0. 22:22:10.808 WARN o.a.k.c.QuorumController - [QuorumController id=0] Performing controller activation. The metadata log appears to be empty. Appending 6 bootstrap record(s) in metadata transaction at metadata.version 4.3-IV0 from bootstrap source 'testkit'. ============================================================================== 1. Idempotence is on by default (Kafka 3.0+) -- verified from the broker ============================================================================== -- What the client library says the defaults are ----------------------------- enable.idempotence : true acks : all -- Sending 3 records with a completely default producer ---------------------- -- Asking the BROKER what producer state it is tracking for idem-default ----- producerId=0 producerEpoch=0 lastSequence=2 producer states tracked : 1 >> The broker holds a real producer ID and the last sequence number it accepted. >> That IS the de-duplication state, and you configured nothing to get it. ============================================================================== 2. acks=1 silently disables idempotence -- no exception, no warning ============================================================================== -- Constructing a producer with ONLY acks=1 ---------------------------------- props: acks=1 (enable.idempotence left at its default of true) producer constructed successfully -- note that nothing complained -- Asking the broker for producer state on idem-acks1 ------------------------ producer states tracked : 0 >> Three records were written, and the broker is tracking NOTHING for them. >> Kafka resolved acks=1 against enable.idempotence=true by turning >> idempotence off, and said nothing at all. Every record this producer >> sends can now be duplicated by an ordinary retry. ============================================================================== 3. acks=1 + enable.idempotence=true is rejected at construction ============================================================================== -- Same acks=1, but this time idempotence is requested explicitly ------------ ConfigException: Must set acks to all in order to use the idempotent producer. Otherwise we cannot guarantee idempotence. >> Ask for it explicitly and Kafka refuses. Leave it implicit and Kafka >> downgrades you. So setting enable.idempotence=true explicitly is worth >> doing purely as an assertion: it turns a silent downgrade into a >> startup failure, which is the failure you want. ============================================================================== 4. The other constraints, and which ones fail loudly ============================================================================== -- max.in.flight.requests.per.connection = 6 --------------------------------- rejected: To use the idempotent producer, max.in.flight.requests.per.connection must be set to at most 5. Current value is 6. -- retries = 0 --------------------------------------------------------------- rejected: Must set retries to non-zero when using the idempotent producer. -- max.in.flight = 5 with idempotence -- allowed ----------------------------- constructed successfully >> Idempotence requires acks=all, retries>0 and max.in.flight<=5. >> Five in flight is safe because the broker can reorder within that window >> using sequence numbers. WITHOUT idempotence, max.in.flight>1 silently >> breaks ORDERING on retry: a retried batch can land after a later one. [INFO] Tests run: 4, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 0.723 s -- in com.ankurm.kafka.eos._01_idempotence.IdempotentProducerTest [INFO] Running com.ankurm.kafka.eos._02_transactions.TransactionMarkerOffsetTest Bootstrap metadata: BootstrapMetadata(records=[ApiMessageAndVersion(FeatureLevelRecord(name='metadata.version', featureLevel=30) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='eligible.leader.replicas.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='group.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='share.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='streams.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='transaction.version', featureLevel=2) at version 0)], metadataVersionLevel=30, source=format command) Formatting metadata directory C:\Users\Ankur\AppData\Local\Temp\kafka-5226944134061393373\combined_0_0 with metadata.version 4.3-IV0. 22:22:11.504 WARN o.a.k.c.QuorumController - [QuorumController id=0] Performing controller activation. The metadata log appears to be empty. Appending 6 bootstrap record(s) in metadata transaction at metadata.version 4.3-IV0 from bootstrap source 'testkit'. ============================================================================== 1. A committed transaction burns one extra offset per partition ============================================================================== -- Before: empty topic ------------------------------------------------------- endOffset = 0 -- Sending 5 records inside ONE transaction, then committing ----------------- sent record 0 -> offset 0 sent record 1 -> offset 1 sent record 2 -> offset 2 sent record 3 -> offset 3 sent record 4 -> offset 4 -- After commit -------------------------------------------------------------- records sent : 5 endOffset : 6 <-- 5 records occupy 0..4, the commit marker takes 5 -- What a read_committed consumer actually receives -------------------------- 22:22:16.224 WARN o.a.k.c.c.i.ConsumerCoordinator - [Consumer clientId=consumer-g-committed-1-4, groupId=g-committed-1] Offset commit failed on partition markers-demo-0 at offset 6: This is not the correct coordinator. 22:22:16.442 WARN o.a.k.c.c.i.ConsumerCoordinator - [Consumer clientId=consumer-g-committed-1-4, groupId=g-committed-1] Offset commit failed on partition markers-demo-0 at offset 6: This is not the correct coordinator. 22:22:16.753 WARN o.a.k.c.c.i.ConsumerCoordinator - [Consumer clientId=consumer-g-committed-1-4, groupId=g-committed-1] Offset commit failed on partition markers-demo-0 at offset 6: This is not the correct coordinator. offset=0 value=committed-0 offset=1 value=committed-1 offset=2 value=committed-2 offset=3 value=committed-3 offset=4 value=committed-4 records delivered : 5 >> 5 records were delivered, the last at offset 4, but the end offset is 6. >> The gap at offset 5 is the commit marker. It exists, it is durable, and >> no consumer will ever see it. ============================================================================== 2. An aborted transaction still consumes offsets and disk ============================================================================== endOffset before : 6 -- Sending 3 records inside a transaction, then ABORTING --------------------- sent record 0 -> offset 6 sent record 1 -> offset 7 sent record 2 -> offset 8 abortTransaction() called -- After abort --------------------------------------------------------------- endOffset after : 10 (+4 for 3 aborted records and 1 abort marker) -- read_committed consumer --------------------------------------------------- delivered : 5 records, values=[committed-0, committed-1, committed-2, committed-3, committed-4] -- read_uncommitted consumer -- the SAME partition --------------------------- delivered : 8 records, values=[committed-0, committed-1, committed-2, committed-3, committed-4, aborted-0, aborted-1, aborted-2] >> The aborted records are physically in the log with permanent offsets. >> isolation.level is a CONSUMER-side filter, not a broker-side delete. >> A default consumer (read_uncommitted) reads data you deliberately aborted. ============================================================================== 3. Lag is NOT inflated -- but counting records by offset arithmetic is broken ============================================================================== -- A read_committed consumer drains the partition, then commits -------------- -- The lag calculation everyone worries about -------------------------------- partition end offset : 10 consumer committed offset : 10 lag (end - committed) : 0 >> Lag is ZERO. The consumer's position advances over markers and aborted >> records just as it advances over delivered ones, so the standard lag >> formula stays correct. The common claim that transactions inflate lag >> does not survive contact with a broker. -- The calculation that IS broken -------------------------------------------- beginningOffset : 0 endOffset : 10 end - beginning : 10 <-- looks like the record count records actually retrievable: 5 offset of last record : 4 overstatement : 100% >> Offset span says 10 records. Only 5 can ever be read -- a 100% error, >> from 2 markers and 3 aborted records. Anything that counts events by >> subtracting offsets -- retention math, throughput dashboards, tests that >> assert on offsets -- is wrong by however much your abort rate happens to be. [INFO] Tests run: 3, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 15.20 s -- in com.ankurm.kafka.eos._02_transactions.TransactionMarkerOffsetTest [INFO] Running com.ankurm.kafka.eos._03_read_process_write.ReadProcessWriteTest . ____ _ __ _ _ /\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \ ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \ \\/ ___)| |_)| | | | | || (_| | ) ) ) ) ' |____| .__|_| |_|_| |_\__, | / / / / =========|_|==============|___/=/_/_/_/ :: Spring Boot :: (v4.1.0) Bootstrap metadata: BootstrapMetadata(records=[ApiMessageAndVersion(FeatureLevelRecord(name='metadata.version', featureLevel=30) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='eligible.leader.replicas.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='group.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='share.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='streams.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='transaction.version', featureLevel=2) at version 0)], metadataVersionLevel=30, source=format command) Formatting metadata directory C:\Users\Ankur\AppData\Local\Temp\kafka-7564387337135394901\combined_0_0 with metadata.version 4.3-IV0. 22:22:27.088 WARN o.a.k.c.QuorumController - [QuorumController id=0] Performing controller activation. The metadata log appears to be empty. Appending 6 bootstrap record(s) in metadata transaction at metadata.version 4.3-IV0 from bootstrap source 'testkit'. 22:22:27.249 INFO c.a.k.e._.ReadProcessWriteTest - Starting ReadProcessWriteTest using Java 25.0.3 with PID 38092 (started by Ankur in C:\Users\Ankur\ankurm-blog-tools\projects\spring-kafka-exactly-once-boot4) 22:22:27.250 INFO c.a.k.e._.ReadProcessWriteTest - No active profile set, falling back to 1 default profile: "default" 22:22:30.619 INFO c.a.k.e._.ReadProcessWriteTest - Started ReadProcessWriteTest in 3.874 seconds (process running for 23.722) ============================================================================== 1. Read-process-write, happy path ============================================================================== -- Publishing 3 orders to orders.incoming ------------------------------------ 22:22:30.756 WARN k.s.DefaultAutoTopicCreationManager - Auto topic creation failed for __transaction_state with error 'TOPIC_ALREADY_EXISTS': Topic '__transaction_state' already exists. 22:22:30.855 WARN k.s.DefaultAutoTopicCreationManager - Auto topic creation failed for __transaction_state with error 'TOPIC_ALREADY_EXISTS': Topic '__transaction_state' already exists. 22:22:30.957 WARN k.s.DefaultAutoTopicCreationManager - Auto topic creation failed for __transaction_state with error 'TOPIC_ALREADY_EXISTS': Topic '__transaction_state' already exists. 22:22:31.059 WARN k.s.DefaultAutoTopicCreationManager - Auto topic creation failed for __transaction_state with error 'TOPIC_ALREADY_EXISTS': Topic '__transaction_state' already exists. 22:22:31.162 WARN k.s.DefaultAutoTopicCreationManager - Auto topic creation failed for __transaction_state with error 'TOPIC_ALREADY_EXISTS': Topic '__transaction_state' already exists. 22:22:31.263 WARN k.s.DefaultAutoTopicCreationManager - Auto topic creation failed for __transaction_state with error 'TOPIC_ALREADY_EXISTS': Topic '__transaction_state' already exists. sent: alpha sent: beta sent: gamma 22:22:32.224 INFO c.a.k.e.o.OrderProcessor - processing 'alpha' (invocation #1) 22:22:32.351 INFO c.a.k.e.o.OrderProcessor - processing 'beta' (invocation #2) 22:22:32.462 INFO c.a.k.e.o.OrderProcessor - processing 'gamma' (invocation #3) -- Listener invocations ------------------------------------------------------ processed: alpha processed: beta processed: gamma -- What a read_committed consumer sees on orders.validated ------------------- offset=0 value=ALPHA offset=2 value=BETA offset=4 value=GAMMA >> 3 in, 3 out. The listener body contains no transaction code at all -- >> the guarantee comes entirely from spring.kafka.producer.transaction-id-prefix. ============================================================================== 2. Failure AFTER the send: processed twice, delivered once ============================================================================== -- Poisoning 'delta' so its first processing attempt throws ------------------ sent: delta -- Waiting for the listener to be invoked more than once --------------------- 22:22:35.108 INFO c.a.k.e.o.OrderProcessor - processing 'delta' (invocation #1) 22:22:35.111 ERROR o.s.k.s.LoggingProducerListener - Exception thrown when sending a message with key='null' and payload='DELTA' to topic orders.validated: org.apache.kafka.common.errors.TransactionAbortedException: Failing batch since transaction was aborted 22:22:35.116 ERROR o.s.k.l.KafkaMessageListenerContainer - Transaction rolled back org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public void com.ankurm.kafka.eos.orders.OrderProcessor.process(java.lang.String)' threw exception at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.decorateException(KafkaMessageListenerContainer.java:3160) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:3031) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:2997) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:2907) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.lambda$invokeInTransaction$0(KafkaMessageListenerContainer.java:2700) at org.springframework.transaction.support.TransactionOperations.lambda$executeWithoutResult$0(TransactionOperations.java:68) at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:137) at org.springframework.transaction.support.TransactionOperations.executeWithoutResult(TransactionOperations.java:67) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeInTransaction(KafkaMessageListenerContainer.java:2696) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListenerInTx(KafkaMessageListenerContainer.java:2667) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:2640) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:2274) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeIfHaveRecords(KafkaMessageListenerContainer.java:1572) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1504) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1373) at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1825) at java.base/java.lang.Thread.run(Thread.java:1474) Suppressed: org.springframework.kafka.listener.ListenerExecutionFailedException: Restored Stack Trace at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:514) at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invoke(MessagingMessageListenerAdapter.java:426) at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:92) at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:52) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:3018) Caused by: java.lang.IllegalStateException: simulated processing failure for 'delta' at com.ankurm.kafka.eos.orders.OrderProcessor.process(OrderProcessor.java:82) at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104) at java.base/java.lang.reflect.Method.invoke(Method.java:565) at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:168) at org.springframework.kafka.listener.adapter.KotlinAwareInvocableHandlerMethod.doInvoke(KotlinAwareInvocableHandlerMethod.java:48) at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:119) at org.springframework.kafka.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:80) at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:489) at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invoke(MessagingMessageListenerAdapter.java:426) at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:92) at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:52) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:3018) ... 15 common frames omitted 22:22:35.613 INFO c.a.k.e.o.OrderProcessor - processing 'delta' (invocation #2) times the listener ran for 'delta' : 2 -- What a read_committed consumer sees --------------------------------------- DELTA records visible : 1 -- What a read_uncommitted consumer sees on the SAME topic ------------------- DELTA records physically in the log : 1 >> The listener ran 2 times and a read_committed consumer sees exactly 1. >> >> This is the whole idea: exactly-once is a property of what is COMMITTED, >> not of how many times your code ran. Any side effect your listener performs >> outside Kafka -- a DB write, an HTTP call, an email -- happened 2 times. -- A detail worth noticing in the number above ------------------------------- read_uncommitted sees 1 DELTA record(s), not 2 >> The rolled-back send never reached the broker at all. KafkaTemplate.send() >> is asynchronous: the record sat in the producer's accumulator, and >> abortTransaction() discarded the un-flushed batch instead of writing it. >> >> So aborted records reach the log only if they were already flushed -- which >> happens under load, on large batches, or when you block on the send future. >> TransactionMarkerOffsetTest forces exactly that case by calling get() on >> each send, and there the aborted records DO occupy permanent offsets. >> Do not rely on either behaviour: rely on read_committed. [INFO] Tests run: 2, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 15.17 s -- in com.ankurm.kafka.eos._03_read_process_write.ReadProcessWriteTest [INFO] Running com.ankurm.kafka.eos._04_fencing.ZombieFencingTest Bootstrap metadata: BootstrapMetadata(records=[ApiMessageAndVersion(FeatureLevelRecord(name='metadata.version', featureLevel=30) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='eligible.leader.replicas.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='group.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='share.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='streams.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='transaction.version', featureLevel=2) at version 0)], metadataVersionLevel=30, source=format command) Formatting metadata directory C:\Users\Ankur\AppData\Local\Temp\kafka-2096764983482195835\combined_0_0 with metadata.version 4.3-IV0. 22:22:42.230 WARN o.a.k.c.QuorumController - [QuorumController id=0] Performing controller activation. The metadata log appears to be empty. Appending 6 bootstrap record(s) in metadata transaction at metadata.version 4.3-IV0 from bootstrap source 'testkit'. ============================================================================== 1. A second producer with the same transactional.id fences the first ============================================================================== -- Producer A ('the zombie') starts a transaction and sends ------------------ transactional.id = 'shared-tx-id' record sent, transaction still OPEN -- Producer B starts with the SAME transactional.id -------------------------- initTransactions() returned -- the coordinator has bumped the epoch any in-flight transaction under the old epoch is now aborted -- Producer A tries to commit ------------------------------------------------ REJECTED with ProducerFencedException message: There is a newer producer with the same transactionalId which fences the current one. -- Producer B commits normally ----------------------------------------------- committed successfully >> The zombie could not commit. Its work was discarded by the broker, >> not by any code you wrote. That is the entire value of transactional.id: >> a durable identity the coordinator can fence on. >> >> Note the exception type -- on Kafka 4.x with KIP-890 (transaction.version=2) >> the epoch is bumped every transaction, so you may see >> InvalidProducerEpochException where older articles promise >> ProducerFencedException. Catch both, or catch neither and let the >> container restart the producer, which is what Spring does. ============================================================================== 2. A random transactional.id per start-up defeats fencing entirely ============================================================================== -- Producer A with a UUID-based transactional.id ----------------------------- transactional.id = 'processor-9900b3b5-0861-4908-93a5-4d4b64c83ca7' -- Producer B restarts and generates a NEW UUID ------------------------------ transactional.id = 'processor-1dd8cdf3-8ddf-4d3d-b28b-152284c558eb' B committed -- Now the 'zombie' A commits -- does anything stop it? ---------------------- A committed successfully. Nothing was fenced. >> Both producers committed. The coordinator saw two unrelated identities >> and had no reason to fence either one. >> >> This is the most damaging Kafka transactions misconfiguration, because >> everything appears to work: transactions commit, tests pass, and you have >> exactly the duplicate-processing problem you turned transactions on to >> prevent. The transactional.id must be STABLE across restarts and tied to >> the partition set an instance owns -- not to the process lifetime. [INFO] Tests run: 2, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 1.944 s -- in com.ankurm.kafka.eos._04_fencing.ZombieFencingTest [INFO] Running com.ankurm.kafka.eos._05_database.DatabaseBoundaryTest . ____ _ __ _ _ /\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \ ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \ \\/ ___)| |_)| | | | | || (_| | ) ) ) ) ' |____| .__|_| |_|_| |_\__, | / / / / =========|_|==============|___/=/_/_/_/ :: Spring Boot :: (v4.1.0) Bootstrap metadata: BootstrapMetadata(records=[ApiMessageAndVersion(FeatureLevelRecord(name='metadata.version', featureLevel=30) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='eligible.leader.replicas.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='group.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='share.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='streams.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='transaction.version', featureLevel=2) at version 0)], metadataVersionLevel=30, source=format command) Formatting metadata directory C:\Users\Ankur\AppData\Local\Temp\kafka-17444936701909163355\combined_0_0 with metadata.version 4.3-IV0. 22:22:43.855 WARN o.a.k.c.QuorumController - [QuorumController id=0] Performing controller activation. The metadata log appears to be empty. Appending 6 bootstrap record(s) in metadata transaction at metadata.version 4.3-IV0 from bootstrap source 'testkit'. 22:22:44.000 INFO c.a.k.e._.DatabaseBoundaryTest - Starting DatabaseBoundaryTest using Java 25.0.3 with PID 38092 (started by Ankur in C:\Users\Ankur\ankurm-blog-tools\projects\spring-kafka-exactly-once-boot4) 22:22:44.000 INFO c.a.k.e._.DatabaseBoundaryTest - No active profile set, falling back to 1 default profile: "default" 22:22:44.452 WARN o.a.k.clients.NetworkClient - [Consumer clientId=consumer-order-processor-15, groupId=order-processor] The metadata response from the cluster reported a recoverable issue with correlation id 2 : {orders.incoming=UNKNOWN_TOPIC_OR_PARTITION} 22:22:44.454 INFO c.a.k.e._.DatabaseBoundaryTest - Started DatabaseBoundaryTest in 0.704 seconds (process running for 37.556) 22:22:44.574 WARN o.a.k.clients.NetworkClient - [Consumer clientId=consumer-order-processor-15, groupId=order-processor] The metadata response from the cluster reported a recoverable issue with correlation id 5 : {orders.incoming=UNKNOWN_TOPIC_OR_PARTITION} 22:22:44.575 WARN k.s.DefaultAutoTopicCreationManager - Auto topic creation failed for orders.incoming with error 'TOPIC_ALREADY_EXISTS': Topic 'orders.incoming' already exists. ============================================================================== 1. The problem: one row per ATTEMPT, not per message ============================================================================== -- Sending one message that will fail on its first processing attempt -------- 22:22:44.765 WARN o.a.k.clients.NetworkClient - [Consumer clientId=consumer-order-processor-15, groupId=order-processor] The metadata response from the cluster reported a recoverable issue with correlation id 15 : {orders.incoming=UNKNOWN_TOPIC_OR_PARTITION} 22:22:44.765 WARN k.s.DefaultAutoTopicCreationManager - Auto topic creation failed for orders.incoming with error 'TOPIC_ALREADY_EXISTS': Topic 'orders.incoming' already exists. 22:22:44.797 WARN k.s.DefaultAutoTopicCreationManager - Auto topic creation failed for __transaction_state with error 'TOPIC_ALREADY_EXISTS': Topic '__transaction_state' already exists. 22:22:44.900 WARN k.s.DefaultAutoTopicCreationManager - Auto topic creation failed for __transaction_state with error 'TOPIC_ALREADY_EXISTS': Topic '__transaction_state' already exists. 22:22:45.002 WARN k.s.DefaultAutoTopicCreationManager - Auto topic creation failed for __transaction_state with error 'TOPIC_ALREADY_EXISTS': Topic '__transaction_state' already exists. 22:22:45.104 WARN k.s.DefaultAutoTopicCreationManager - Auto topic creation failed for __transaction_state with error 'TOPIC_ALREADY_EXISTS': Topic '__transaction_state' already exists. 22:22:45.975 INFO c.a.k.e._.DatabaseBoundaryTest$Listeners$NaiveDbListener - naive listener wrote a row for 'order-1' 22:22:45.975 ERROR o.s.k.l.KafkaMessageListenerContainer - Transaction rolled back org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public void com.ankurm.kafka.eos._05_database.DatabaseBoundaryTest$Listeners$NaiveDbListener.onMessage(java.lang.String)' threw exception at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.decorateException(KafkaMessageListenerContainer.java:3160) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:3031) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:2997) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:2907) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.lambda$invokeInTransaction$0(KafkaMessageListenerContainer.java:2700) at org.springframework.transaction.support.TransactionOperations.lambda$executeWithoutResult$0(TransactionOperations.java:68) at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:137) at org.springframework.transaction.support.TransactionOperations.executeWithoutResult(TransactionOperations.java:67) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeInTransaction(KafkaMessageListenerContainer.java:2696) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListenerInTx(KafkaMessageListenerContainer.java:2667) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:2640) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:2274) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeIfHaveRecords(KafkaMessageListenerContainer.java:1572) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1504) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1373) at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1825) at java.base/java.lang.Thread.run(Thread.java:1474) Suppressed: org.springframework.kafka.listener.ListenerExecutionFailedException: Restored Stack Trace at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:514) at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invoke(MessagingMessageListenerAdapter.java:426) at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:92) at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:52) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:3018) Caused by: java.lang.IllegalStateException: simulated failure after the DB write for order-1 at com.ankurm.kafka.eos._05_database.DatabaseBoundaryTest$Listeners$Base.maybeFail(DatabaseBoundaryTest.java:209) at com.ankurm.kafka.eos._05_database.DatabaseBoundaryTest$Listeners$NaiveDbListener.onMessage(DatabaseBoundaryTest.java:243) at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104) at java.base/java.lang.reflect.Method.invoke(Method.java:565) at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:168) at org.springframework.kafka.listener.adapter.KotlinAwareInvocableHandlerMethod.doInvoke(KotlinAwareInvocableHandlerMethod.java:48) at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:119) at org.springframework.kafka.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:80) at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:489) at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invoke(MessagingMessageListenerAdapter.java:426) at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:92) at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:52) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:3018) ... 15 common frames omitted 22:22:46.438 INFO c.a.k.e._.DatabaseBoundaryTest$Listeners$NaiveDbListener - naive listener wrote a row for 'order-1' listener attempts for 'order-1' : 2 rows in the database : 2 id=1 dedupeKey=naive-1 id=2 dedupeKey=naive-2 >> The Kafka transaction rolled back correctly. The database row did not, >> because it was never in the Kafka transaction and never could have been. >> >> Note that nothing here is misconfigured. transaction-id-prefix is set, >> isolation.level is read_committed, the Kafka side is exactly-once. The >> database still has 2 rows for 1 message. ============================================================================== 2. The fix: make the side effect idempotent, not the delivery ============================================================================== -- Same failure, but the listener dedupes on topic-partition-offset ---------- 22:22:48.940 INFO c.a.k.e._.DatabaseBoundaryTest$Listeners$IdempotentDbListener - idempotent listener wrote a row for 'order-2' as db.idempotent-0-0 22:22:48.941 ERROR o.s.k.l.KafkaMessageListenerContainer - Transaction rolled back org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method 'public void com.ankurm.kafka.eos._05_database.DatabaseBoundaryTest$Listeners$IdempotentDbListener.onMessage(java.lang.String,java.lang.String,int,long)' threw exception at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.decorateException(KafkaMessageListenerContainer.java:3160) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:3031) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:2997) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:2907) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.lambda$invokeInTransaction$0(KafkaMessageListenerContainer.java:2700) at org.springframework.transaction.support.TransactionOperations.lambda$executeWithoutResult$0(TransactionOperations.java:68) at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:137) at org.springframework.transaction.support.TransactionOperations.executeWithoutResult(TransactionOperations.java:67) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeInTransaction(KafkaMessageListenerContainer.java:2696) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListenerInTx(KafkaMessageListenerContainer.java:2667) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:2640) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:2274) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeIfHaveRecords(KafkaMessageListenerContainer.java:1572) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1504) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1373) at java.base/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1825) at java.base/java.lang.Thread.run(Thread.java:1474) Suppressed: org.springframework.kafka.listener.ListenerExecutionFailedException: Restored Stack Trace at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:514) at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invoke(MessagingMessageListenerAdapter.java:426) at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:92) at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:52) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:3018) Caused by: java.lang.IllegalStateException: simulated failure after the DB write for order-2 at com.ankurm.kafka.eos._05_database.DatabaseBoundaryTest$Listeners$Base.maybeFail(DatabaseBoundaryTest.java:209) at com.ankurm.kafka.eos._05_database.DatabaseBoundaryTest$Listeners$IdempotentDbListener.onMessage(DatabaseBoundaryTest.java:275) at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104) at java.base/java.lang.reflect.Method.invoke(Method.java:565) at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:168) at org.springframework.kafka.listener.adapter.KotlinAwareInvocableHandlerMethod.doInvoke(KotlinAwareInvocableHandlerMethod.java:48) at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:119) at org.springframework.kafka.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:80) at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:489) at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invoke(MessagingMessageListenerAdapter.java:426) at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:92) at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:52) at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:3018) ... 15 common frames omitted 22:22:49.423 INFO c.a.k.e._.DatabaseBoundaryTest$Listeners$IdempotentDbListener - idempotent listener SKIPPED 'order-2' -- db.idempotent-0-0 already processed listener attempts for 'order-2' : 2 rows in the database : 1 id=3 dedupeKey=db.idempotent-0-0 >> The listener ran 2 times and wrote 1 row. >> >> The dedupe key is 'topic-partition-offset', which Kafka guarantees is >> unique and stable. A UNIQUE constraint on that column turns the second >> attempt into a no-op -- and, importantly, would still do so if the >> duplicate came from a different process, a rebalance, or a replay. >> >> This is the whole design rule: exactly-once processing is achieved by >> idempotent side effects plus at-least-once delivery. Kafka transactions >> give you the second half. You have to build the first half yourself. [INFO] Tests run: 2, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 7.701 s -- in com.ankurm.kafka.eos._05_database.DatabaseBoundaryTest [INFO] Running com.ankurm.kafka.eos._06_performance.TransactionCostTest Bootstrap metadata: BootstrapMetadata(records=[ApiMessageAndVersion(FeatureLevelRecord(name='metadata.version', featureLevel=30) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='eligible.leader.replicas.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='group.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='share.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='streams.version', featureLevel=1) at version 0), ApiMessageAndVersion(FeatureLevelRecord(name='transaction.version', featureLevel=2) at version 0)], metadataVersionLevel=30, source=format command) Formatting metadata directory C:\Users\Ankur\AppData\Local\Temp\kafka-8432095967619283703\combined_0_0 with metadata.version 4.3-IV0. 22:22:52.001 WARN o.a.k.c.QuorumController - [QuorumController id=0] Performing controller activation. The metadata log appears to be empty. Appending 6 bootstrap record(s) in metadata transaction at metadata.version 4.3-IV0 from bootstrap source 'testkit'. ============================================================================== The cost of a transaction is paid per COMMIT, not per record ============================================================================== Sending 10,000 records of 256 bytes in every configuration. -- warm-up (discarded) ------------------------------------------------------- done -- 5 interleaved rounds of 10,000 records (round 0 discarded) ---------------- round 0 (discarded): baseline=311,056 1/tx=349 10/tx=8,032 100/tx=66,664 1000/tx=344,474 10000/tx=641,264 round 1: baseline=628,417 1/tx=765 10/tx=8,452 100/tx=61,437 1000/tx=434,245 10000/tx=836,617 round 2: baseline=801,880 1/tx=685 10/tx=8,499 100/tx=78,770 1000/tx=216,065 10000/tx=878,071 round 3: baseline=760,537 1/tx=718 10/tx=7,467 100/tx=104,725 1000/tx=235,578 10000/tx=942,649 round 4: baseline=833,848 1/tx=711 10/tx=8,357 100/tx=71,036 1000/tx=471,918 10000/tx=897,892 round 5: baseline=818,940 1/tx=692 10/tx=8,491 100/tx=97,746 1000/tx=330,969 10000/tx=657,960 -- measurement stability ----------------------------------------------------- no transaction (idempotent) median=801,880 spread=+26% 1 records/transaction median=711 spread=+11% 10 records/transaction median=8,452 spread=+12% 100 records/transaction median=78,770 spread=+55% 1,000 records/transaction median=330,969 spread=+77% 10,000 records/transaction median=878,071 spread=+32% baseline first counted round : 628,417 records/s baseline last counted round : 818,940 records/s baseline drift : +30.3% -- summary (median of 5 rounds) ---------------------------------------------- configuration records/s vs baseline no transaction (idempotent) 801,880 100% 1 records/transaction 711 0% 10 records/transaction 8,452 1% 100 records/transaction 78,770 10% 1,000 records/transaction 330,969 41% 10,000 records/transaction 878,071 110% >> Records per transaction, and what it buys: >> 1 -> 711 rec/s >> 10 -> 8,452 rec/s (11.9x) >> 100 -> 78,770 rec/s (9.3x) >> 1,000 -> 330,969 rec/s (4.2x) >> >> Each 10x increase in transaction size buys close to 10x throughput while >> commit cost dominates, then flattens as the per-record cost takes over. >> One record per transaction is roughly 111x slower than a hundred. >> >> (The non-transactional baseline is quoted at 801,880 rec/s but varied by >> 26% across rounds, so ratios against it are indicative only. The >> comparisons BETWEEN transaction sizes are the stable, meaningful ones.) >> >> The lesson is not 'transactions are slow'. It is that a commit is a fixed >> cost, so the number of records you put inside one is the setting that >> matters. In Spring, that number is the size of the batch returned by >> poll(), which you control with max.poll.records -- NOT with anything >> named 'transaction'. Raising max.poll.records is the single most >> effective throughput fix for a transactional listener, at the price of >> redoing more work when one record in the batch fails. [INFO] Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 112.4 s -- in com.ankurm.kafka.eos._06_performance.TransactionCostTest [INFO] [INFO] Results: [INFO] [INFO] Tests run: 15, Failures: 0, Errors: 0, Skipped: 0 [INFO] [INFO] ------------------------------------------------------------------------ [INFO] BUILD SUCCESS [INFO] ------------------------------------------------------------------------ [INFO] Total time: 02:46 min [INFO] Finished at: 2026-07-31T22:24:48+05:30 [INFO] ------------------------------------------------------------------------