KAFKA-20716: Document unclean leader election transaction limitations by lh0156 · Pull Request #22937 · apache/kafka

lh0156 · 2026-07-26T09:33:14Z

Summary

Document that enabling unclean leader election on transactional topics
is incompatible with exactly-once semantics.
Explain how a missing transaction marker can leave read_committed
consumers stuck at the last stable offset.
Point operators to kafka-transactions.sh find-hanging and abort
for recovery, with an explicit warning to verify the transaction and
partition first.
Preserve the existing KRaft operational note and fix its missing word
boundary.
Add deterministic end-to-end coverage that reproduces the
stale-follower case and verifies the read_committed result after an
unclean election.

Why

KAFKA-20716 reports
that an unclean election can leave a partition with transactional data
but without its COMMIT/ABORT marker. Maintainers confirmed that unclean
leader election cannot guarantee EOS, asked for this limitation to be
documented clearly, and requested an end-to-end test.

The regression test initializes the transaction coordinator first,
places the data partition leader on that broker, waits until the
follower has the transactional record, stops the follower before the
marker is committed, and then performs an unclean election. It verifies
that a read_committed consumer returns only the earlier
non-transactional record and remains at offset 1.

This change does not alter broker behavior or introduce automatic
transaction reconciliation.

Validation

./gradlew :server:test --tests org.apache.kafka.server.UncleanLeaderElectionTest --no-build-cache --console=plain
./gradlew :server:test --tests org.apache.kafka.server.UncleanLeaderElectionTest.testUncleanLeaderElectionCanLeaveReadCommittedConsumerAtLastStableOffset --no-build-cache --console=plain
./gradlew :server:spotlessCheck :server:checkstyleMain :server:checkstyleTest :server:spotbugsMain
./gradlew :clients:test :clients:spotlessCheck :clients:checkstyleMain :core:genTopicConfigDocs :core:siteDocsTar --no-build-cache
./gradlew :core:genTopicConfigDocs :clients:spotlessCheck :clients:checkstyleMain :clients:spotbugsMain --no-build-cache
git diff --check

The new test passed in both Raft-Isolated and Raft-Combined cluster
variants. The generated docs/generated/topic_config.html was inspected
to verify the new text and HTML rendering.

Fixes KAFKA-20716

Reviewers: Luke Chen showuon@gmail.com

showuon · 2026-08-03T04:07:24Z

+        "If this occurs, use <code>kafka-transactions.sh find-hanging</code> to identify the affected transaction and " +
+        "<code>kafka-transactions.sh abort</code> to recover it. Verify the transaction and partition before aborting " +
+        "it, since an unclean election may already have caused data loss." +
+        "<p>Note: In KRaft mode, when enabling this config dynamically, it needs to wait for the unclean leader election " +


I'm curious, are you sure the kafka-transactions.sh find-hanging and abort can help on the situation? Have you tested it?

I verified the command behavior and updated the documentation in ff259c2.

The guidance is conditional: find-hanging scans DescribeProducers and transaction metadata to identify open producer transactions whose partition marker is missing. If the producer state is still available, abort --topic --partition --start-offset validates that open transaction and issues the abort. It cannot restore records or markers that were lost by the unclean election, so the updated text calls that out explicitly.

The existing TransactionsCommandTest coverage passed, including testFindHangingNoMappedTransactionalId and the broker 3.0+ start-offset abort path. I have not claimed an end-to-end unclean-election reproduction.

It'd be better we did a end-to-end unclean-election test. WDYT?

Thanks, I agree that an end-to-end unclean-election test would be valuable. I kept this PR documentation-only because a reliable test would need to control the election race, reproduce a missing transaction marker, and verify the read_committed behavior; that is a broader and timing-sensitive integration test rather than validation of the documentation change. I would be happy to address that as a separate follow-up if the maintainers want it.

Implemented the requested end-to-end test in UncleanLeaderElectionTest. It aligns the data leader with the transaction coordinator, waits for the follower to replicate the transactional record, stops the follower before the marker commit, then verifies after unclean failover that a read_committed consumer returns only the earlier record and remains at offset 1. The full test class passes in both Raft-Isolated and Raft-Combined variants.

Agreed. I added UncleanLeaderElectionTest.testUncleanLeaderElectionCanLeaveReadCommittedConsumerAtLastStableOffset in commit 6dab67ed22. The test reproduces the missing transaction-marker scenario with a transactional producer, stops the follower before the marker is replicated, performs an unclean election, and verifies that a read_committed consumer stops at the last stable offset. It passes for both Raft-Isolated and Raft-Combined with ./gradlew :server:test --tests org.apache.kafka.server.UncleanLeaderElectionTest.testUncleanLeaderElectionCanLeaveReadCommittedConsumerAtLastStableOffset --no-build-cache --console=plain.

The documentation now scopes kafka-transactions.sh find-hanging and abort to the case where producer state is still available, explicitly warns that they cannot restore a lost marker or data, and asks operators to verify the transaction and partition before aborting. I did not imply that these commands repair the unclean-election data loss.

Document the required scope options for finding hanging transactions and the producer-state precondition for aborting one. Make clear that the transaction tool cannot restore records or markers lost by an unclean election. Generated-by: OpenAI Codex (GPT-5) Signed-off-by: Yunseop Eom <62834176+lh0156@users.noreply.github.com>

Generated-by: OpenAI Codex (GPT-5) Signed-off-by: Yunseop Eom <62834176+lh0156@users.noreply.github.com>

showuon · 2026-08-04T06:37:18Z

+            KafkaBroker leader = cluster.brokers().get(leaderId);
+            KafkaBroker follower = cluster.brokers().get(followerId);
+
+            produceMessage(cluster, TOPIC, "before");


Let's add a comment here, ex: // 1. produce a non-transactional record

showuon · 2026-08-04T06:38:09Z

+                    "Timed out waiting for the follower to replicate the non-transactional record"
+            );
+
+            producer.beginTransaction();


Let's add a comment here, ex: // produce a transactional record

showuon · 2026-08-04T06:39:22Z

+
+            Map<String, Object> consumerConfig = Map.of(
+                    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers(),
+                    ConsumerConfig.GROUP_ID_CONFIG, "unclean-election-consumer-" + UUID.randomUUID(),
+                    ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName(),
+                    ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName(),
+                    ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest",
+                    ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false,
+                    ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.toString()
+            );
+
+            try (Consumer<String, String> consumer = new KafkaConsumer<>(consumerConfig)) {
+                consumer.assign(List.of(TOPIC_PARTITION));
+                consumer.seekToBeginning(List.of(TOPIC_PARTITION));
+                List<String> values = new ArrayList<>();
+                waitForCondition(
+                        () -> {
+                            for (ConsumerRecord<String, String> record : consumer.poll(Duration.ofMillis(100))) {
+                                values.add(record.value());
+                            }
+                            return !values.isEmpty();
+                        },
+                        DEFAULT_MAX_WAIT_MS,
+                        "Timed out waiting for the read_committed consumer to read the non-transactional record"
+                );
+
+                assertEquals(List.of("before"), values);
+                assertEquals(1, consumer.position(TOPIC_PARTITION));
+                assertTrue(consumer.poll(Duration.ofMillis(500)).isEmpty());
+            }
+        }


Suggested change

Map<String, Object> consumerConfig = Map.of(

ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers(),

ConsumerConfig.GROUP_ID_CONFIG, "unclean-election-consumer-" + UUID.randomUUID(),

ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName(),

ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName(),

ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest",

ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false,

ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.toString()

);

try (Consumer<String, String> consumer = new KafkaConsumer<>(consumerConfig)) {

consumer.assign(List.of(TOPIC_PARTITION));

consumer.seekToBeginning(List.of(TOPIC_PARTITION));

List<String> values = new ArrayList<>();

waitForCondition(

() -> {

for (ConsumerRecord<String, String> record : consumer.poll(Duration.ofMillis(100))) {

values.add(record.value());

}

return !values.isEmpty();

},

DEFAULT_MAX_WAIT_MS,

"Timed out waiting for the read_committed consumer to read the non-transactional record"

);

assertEquals(List.of("before"), values);

assertEquals(1, consumer.position(TOPIC_PARTITION));

assertTrue(consumer.poll(Duration.ofMillis(500)).isEmpty());

}

}

// produce a non-transactional record to the follower node (current leader)

produceMessage(cluster, TOPIC, "after");

Map<String, Object> consumerConfig = Map.of(

ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers(),

ConsumerConfig.GROUP_ID_CONFIG, "unclean-election-consumer-" + UUID.randomUUID(),

ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName(),

ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName(),

ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest",

ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false,

ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.toString()

);

// create a consumer to read committed records

try (Consumer<String, String> consumer = new KafkaConsumer<>(consumerConfig)) {

consumer.assign(List.of(TOPIC_PARTITION));

consumer.seekToBeginning(List.of(TOPIC_PARTITION));

List<String> values = new ArrayList<>();

waitForCondition(

() -> {

for (ConsumerRecord<String, String> record : consumer.poll(Duration.ofMillis(100))) {

values.add(record.value());

}

return !values.isEmpty();

},

DEFAULT_MAX_WAIT_MS,

"Timed out waiting for the read_committed consumer to read the non-transactional record"

);

// verify we can only read the 1st non-txn record because the 2nd txn record is not committed in the log

assertEquals(List.of("before"), values);

assertEquals(1, consumer.position(TOPIC_PARTITION));

assertTrue(consumer.poll(Duration.ofMillis(500)).isEmpty());

}

// try to abort the txn

var describeProducerResult = admin.describeProducers(List.of(TOPIC_PARTITION), new DescribeProducersOptions()).all().get();

// find the hanging txn by describeProducer

List<ProducerState> hangingTxnProducerStates = describeProducerResult.get(TOPIC_PARTITION).activeProducers().stream().filter(ap -> ap.currentTransactionStartOffset().isPresent()).toList();

assertEquals(1, hangingTxnProducerStates.size());

// abort the hanging txn

admin.abortTransaction(new AbortTransactionSpec(

TOPIC_PARTITION,

hangingTxnProducerStates.get(0).producerId(),

(short) hangingTxnProducerStates.get(0).producerEpoch(),

hangingTxnProducerStates.get(0).coordinatorEpoch().orElse(0))).all().get();

// create another consumer to read committed records again

try (Consumer<String, String> consumer = new KafkaConsumer<>(consumerConfig)) {

consumer.assign(List.of(TOPIC_PARTITION));

consumer.seekToBeginning(List.of(TOPIC_PARTITION));

List<String> values = new ArrayList<>();

waitForCondition(

() -> {

for (ConsumerRecord<String, String> record : consumer.poll(Duration.ofMillis(100))) {

values.add(record.value());

}

return !values.isEmpty();

},

DEFAULT_MAX_WAIT_MS,

"Timed out waiting for the read_committed consumer to read the non-transactional record"

);

// verify we can now read the 2nd non-txn record because the txn record is aborted in the log

assertEquals(List.of("before", "after"), values);

assertEquals(4, consumer.position(TOPIC_PARTITION));

assertTrue(consumer.poll(Duration.ofMillis(500)).isEmpty());

}

}

Let's write a 3rd record with non-txn record. So that we can use the consumer to verify the committed read is blocked. Then after aborting the hanging txn, we can then verify it's unblocked.

showuon · 2026-08-04T06:45:19Z

+        "If this occurs, use <code>kafka-transactions.sh find-hanging</code> with <code>--topic</code> or " +
+        "<code>--broker-id</code> to look for an open transaction whose marker is missing. If it reports a transaction " +
+        "whose producer state is still available, use <code>kafka-transactions.sh abort</code> with its topic, partition, " +
+        "and start offset to abort it. These commands cannot restore records or markers lost by an unclean election. " +


Suggested change

"If this occurs, use <code>kafka-transactions.sh find-hanging</code> with <code>--topic</code> or " +

"<code>--broker-id</code> to look for an open transaction whose marker is missing. If it reports a transaction " +

"whose producer state is still available, use <code>kafka-transactions.sh abort</code> with its topic, partition, " +

"and start offset to abort it. These commands cannot restore records or markers lost by an unclean election. " +

"If this occurs, use <code>kafka-transactions.sh find-hanging</code> " +

"to look for an open transaction whose marker is missing. If it reports a transaction " +

"whose producer state is still available, use <code>kafka-transactions.sh abort</code> " +

"to abort it. "

showuon · 2026-08-04T06:45:51Z

+        "Verify the transaction and partition before aborting it, since an unclean election may already have caused " +
+        "data loss." +


Suggested change

"Verify the transaction and partition before aborting it, since an unclean election may already have caused " +

"data loss." +

KAFKA-20716: Document unclean election transaction limitations

79ea1d7

github-actions Bot added triage PRs from the community clients small Small PRs labels Jul 26, 2026

showuon reviewed Aug 3, 2026

View reviewed changes

github-actions Bot removed the triage PRs from the community label Aug 3, 2026

lh0156 added 2 commits August 3, 2026 13:56

KAFKA-20716: Add unclean election transaction coverage

3e78ad5

Generated-by: OpenAI Codex (GPT-5) Signed-off-by: Yunseop Eom <62834176+lh0156@users.noreply.github.com>

github-actions Bot added core Kafka Broker and removed small Small PRs labels Aug 3, 2026

KAFKA-20716: Improve regression test readability

6dab67e

Generated-by: OpenAI Codex (GPT-5) Signed-off-by: Yunseop Eom <62834176+lh0156@users.noreply.github.com>

showuon reviewed Aug 4, 2026

View reviewed changes

Navigation Menu

Uh oh!

KAFKA-20716: Document unclean leader election transaction limitations - #22937

KAFKA-20716: Document unclean leader election transaction limitations#22937
lh0156 wants to merge 4 commits into
apache:trunkfrom
lh0156:agent/KAFKA-20716-document-unclean-election

lh0156 commented Jul 26, 2026 •

edited by github-actions Bot

Loading

Uh oh!

showuon Aug 3, 2026 •

edited

Loading

Uh oh!

lh0156 Aug 3, 2026

Uh oh!

showuon Aug 3, 2026

Uh oh!

lh0156 Aug 3, 2026

Uh oh!

lh0156 Aug 3, 2026

Uh oh!

lh0156 Aug 3, 2026

Uh oh!

showuon Aug 4, 2026

Uh oh!

showuon Aug 4, 2026

Uh oh!

showuon Aug 4, 2026

Uh oh!

showuon Aug 4, 2026

Uh oh!

showuon Aug 4, 2026

Uh oh!

showuon Aug 4, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

		"Verify the transaction and partition before aborting it, since an unclean election may already have caused " +
		"data loss." +

Uh oh!

Conversation

lh0156 commented Jul 26, 2026 • edited by github-actions Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Why

Validation

Uh oh!

showuon Aug 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

lh0156 commented Jul 26, 2026 •

edited by github-actions Bot

Loading

showuon Aug 3, 2026 •

edited

Loading