[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] ------------------------------------------------------------------------
