showuon commented on code in PR #22937:
URL: https://github.com/apache/kafka/pull/22937#discussion_r3710038832
##########
server/src/test/java/org/apache/kafka/server/UncleanLeaderElectionTest.java:
##########
@@ -139,6 +152,126 @@ public void testUncleanLeaderElectionEnabledConsumer()
throws Exception {
testUncleanLeaderElectionEnabled(GroupProtocol.CONSUMER);
}
+ @ClusterTest(
+ serverProperties = {
+ @ClusterConfigProperty(key =
TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key =
TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key =
TransactionLogConfig.TRANSACTIONS_TOPIC_MIN_ISR_CONFIG, value = "1")
+ }
+ )
+ public void
testUncleanLeaderElectionCanLeaveReadCommittedConsumerAtLastStableOffset()
throws Exception {
+ disableEligibleLeaderReplicas();
+
+ String transactionalId = "unclean-election-" + UUID.randomUUID();
+ Map<String, Object> producerConfig = Map.of(
+ ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
cluster.bootstrapServers(),
+ ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName(),
+ ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName(),
+ ProducerConfig.ACKS_CONFIG, "all",
+ ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId
+ );
+
+ try (KafkaProducer<String, String> producer = new
KafkaProducer<>(producerConfig)) {
+ producer.initTransactions();
+
+ var transactionStateTopic =
admin.describeTopics(Set.of(Topic.TRANSACTION_STATE_TOPIC_NAME))
+
.allTopicNames().get().get(Topic.TRANSACTION_STATE_TOPIC_NAME);
+ int transactionPartitionId = Utils.abs(transactionalId.hashCode())
% transactionStateTopic.partitions().size();
+ int transactionCoordinatorId =
transactionStateTopic.partitions().stream()
+ .filter(partition -> partition.partition() ==
transactionPartitionId)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Transaction state
partition is missing"))
+ .leader().id();
+ int followerId = transactionCoordinatorId == BROKER_ID_0 ?
BROKER_ID_1 : BROKER_ID_0;
+
+ NewTopic newTopic = new NewTopic(TOPIC, Map.of(PARTITION_ID,
List.of(transactionCoordinatorId, followerId)))
+ .configs(Map.of(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG,
"1"));
+ admin.createTopics(List.of(newTopic)).all().get();
+ try {
+ admin.electLeaders(ElectionType.PREFERRED,
Set.of(TOPIC_PARTITION)).all().get();
+ } catch (ExecutionException e) {
+ assertInstanceOf(ElectionNotNeededException.class,
e.getCause());
+ }
+
+ int leaderId = awaitLeaderChange(cluster, TOPIC_PARTITION,
Optional.of(transactionCoordinatorId));
+ KafkaBroker leader = cluster.brokers().get(leaderId);
+ KafkaBroker follower = cluster.brokers().get(followerId);
+
+ produceMessage(cluster, TOPIC, "before");
+ waitForCondition(
+ () -> {
+ var log =
follower.replicaManager().localLog(TOPIC_PARTITION);
+ return log.isDefined() && log.get().logEndOffset() >=
1;
+ },
+ DEFAULT_MAX_WAIT_MS,
+ "Timed out waiting for the follower to replicate the
non-transactional record"
+ );
+
+ producer.beginTransaction();
+ producer.send(new ProducerRecord<>(TOPIC, "transactional")).get();
+
+ // The follower must contain the transaction data, but must be
stopped before the marker is committed.
+ waitForCondition(
+ () -> {
+ var log =
follower.replicaManager().localLog(TOPIC_PARTITION);
+ return log.isDefined() && log.get().logEndOffset() >=
2;
+ },
+ DEFAULT_MAX_WAIT_MS,
+ "Timed out waiting for the follower to replicate the
transactional record"
+ );
+ follower.shutdown();
+ follower.awaitShutdown();
+ waitForCondition(
+ () -> {
+ var leaderAndIsr =
leader.replicaManager().metadataCache().getLeaderAndIsr(TOPIC, PARTITION_ID);
+ return leaderAndIsr.isPresent() &&
leaderAndIsr.get().isr().equals(Set.of(leaderId));
+ },
+ DEFAULT_MAX_WAIT_MS,
+ "Timed out waiting for the follower to leave the ISR"
+ );
+
+ // The marker is now written only to the leader. The follower will
retain the open transaction after failover.
+ producer.commitTransaction();
+
+ // Close the producer while its coordinator is still available; no
client cleanup should depend on the old leader.
+ producer.close();
+ leader.shutdown();
+ leader.awaitShutdown();
+ follower.startup();
+ awaitLeaderChange(cluster, TOPIC_PARTITION,
Optional.of(followerId));
+
+ 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());
+ }
+ }
Review Comment:
```suggestion
// 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());
}
}
```
##########
server/src/test/java/org/apache/kafka/server/UncleanLeaderElectionTest.java:
##########
@@ -139,6 +152,126 @@ public void testUncleanLeaderElectionEnabledConsumer()
throws Exception {
testUncleanLeaderElectionEnabled(GroupProtocol.CONSUMER);
}
+ @ClusterTest(
+ serverProperties = {
+ @ClusterConfigProperty(key =
TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key =
TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key =
TransactionLogConfig.TRANSACTIONS_TOPIC_MIN_ISR_CONFIG, value = "1")
+ }
+ )
+ public void
testUncleanLeaderElectionCanLeaveReadCommittedConsumerAtLastStableOffset()
throws Exception {
+ disableEligibleLeaderReplicas();
+
+ String transactionalId = "unclean-election-" + UUID.randomUUID();
+ Map<String, Object> producerConfig = Map.of(
+ ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
cluster.bootstrapServers(),
+ ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName(),
+ ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName(),
+ ProducerConfig.ACKS_CONFIG, "all",
+ ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId
+ );
+
+ try (KafkaProducer<String, String> producer = new
KafkaProducer<>(producerConfig)) {
+ producer.initTransactions();
+
+ var transactionStateTopic =
admin.describeTopics(Set.of(Topic.TRANSACTION_STATE_TOPIC_NAME))
+
.allTopicNames().get().get(Topic.TRANSACTION_STATE_TOPIC_NAME);
+ int transactionPartitionId = Utils.abs(transactionalId.hashCode())
% transactionStateTopic.partitions().size();
+ int transactionCoordinatorId =
transactionStateTopic.partitions().stream()
+ .filter(partition -> partition.partition() ==
transactionPartitionId)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Transaction state
partition is missing"))
+ .leader().id();
+ int followerId = transactionCoordinatorId == BROKER_ID_0 ?
BROKER_ID_1 : BROKER_ID_0;
+
+ NewTopic newTopic = new NewTopic(TOPIC, Map.of(PARTITION_ID,
List.of(transactionCoordinatorId, followerId)))
+ .configs(Map.of(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG,
"1"));
+ admin.createTopics(List.of(newTopic)).all().get();
+ try {
+ admin.electLeaders(ElectionType.PREFERRED,
Set.of(TOPIC_PARTITION)).all().get();
+ } catch (ExecutionException e) {
+ assertInstanceOf(ElectionNotNeededException.class,
e.getCause());
+ }
+
+ int leaderId = awaitLeaderChange(cluster, TOPIC_PARTITION,
Optional.of(transactionCoordinatorId));
+ KafkaBroker leader = cluster.brokers().get(leaderId);
+ KafkaBroker follower = cluster.brokers().get(followerId);
+
+ produceMessage(cluster, TOPIC, "before");
+ waitForCondition(
+ () -> {
+ var log =
follower.replicaManager().localLog(TOPIC_PARTITION);
+ return log.isDefined() && log.get().logEndOffset() >=
1;
+ },
+ DEFAULT_MAX_WAIT_MS,
+ "Timed out waiting for the follower to replicate the
non-transactional record"
+ );
+
+ producer.beginTransaction();
Review Comment:
Let's add a comment here, ex: `// produce a transactional record`
##########
server/src/test/java/org/apache/kafka/server/UncleanLeaderElectionTest.java:
##########
@@ -139,6 +152,126 @@ public void testUncleanLeaderElectionEnabledConsumer()
throws Exception {
testUncleanLeaderElectionEnabled(GroupProtocol.CONSUMER);
}
+ @ClusterTest(
+ serverProperties = {
+ @ClusterConfigProperty(key =
TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key =
TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key =
TransactionLogConfig.TRANSACTIONS_TOPIC_MIN_ISR_CONFIG, value = "1")
+ }
+ )
+ public void
testUncleanLeaderElectionCanLeaveReadCommittedConsumerAtLastStableOffset()
throws Exception {
+ disableEligibleLeaderReplicas();
+
+ String transactionalId = "unclean-election-" + UUID.randomUUID();
+ Map<String, Object> producerConfig = Map.of(
+ ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
cluster.bootstrapServers(),
+ ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName(),
+ ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName(),
+ ProducerConfig.ACKS_CONFIG, "all",
+ ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId
+ );
+
+ try (KafkaProducer<String, String> producer = new
KafkaProducer<>(producerConfig)) {
+ producer.initTransactions();
+
+ var transactionStateTopic =
admin.describeTopics(Set.of(Topic.TRANSACTION_STATE_TOPIC_NAME))
+
.allTopicNames().get().get(Topic.TRANSACTION_STATE_TOPIC_NAME);
+ int transactionPartitionId = Utils.abs(transactionalId.hashCode())
% transactionStateTopic.partitions().size();
+ int transactionCoordinatorId =
transactionStateTopic.partitions().stream()
+ .filter(partition -> partition.partition() ==
transactionPartitionId)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Transaction state
partition is missing"))
+ .leader().id();
+ int followerId = transactionCoordinatorId == BROKER_ID_0 ?
BROKER_ID_1 : BROKER_ID_0;
+
+ NewTopic newTopic = new NewTopic(TOPIC, Map.of(PARTITION_ID,
List.of(transactionCoordinatorId, followerId)))
+ .configs(Map.of(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG,
"1"));
+ admin.createTopics(List.of(newTopic)).all().get();
+ try {
+ admin.electLeaders(ElectionType.PREFERRED,
Set.of(TOPIC_PARTITION)).all().get();
+ } catch (ExecutionException e) {
+ assertInstanceOf(ElectionNotNeededException.class,
e.getCause());
+ }
+
+ int leaderId = awaitLeaderChange(cluster, TOPIC_PARTITION,
Optional.of(transactionCoordinatorId));
+ KafkaBroker leader = cluster.brokers().get(leaderId);
+ KafkaBroker follower = cluster.brokers().get(followerId);
+
+ produceMessage(cluster, TOPIC, "before");
Review Comment:
Let's add a comment here, ex: `// 1. produce a non-transactional record`
##########
clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java:
##########
@@ -199,7 +199,16 @@ public class TopicConfig {
public static final String UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG =
"unclean.leader.election.enable";
public static final String UNCLEAN_LEADER_ELECTION_ENABLE_DOC = "Indicates
whether to enable replicas " +
"not in the ISR set to be elected as leader as a last resort, even
though doing so may result in data " +
- "loss.<p>Note: In KRaft mode, when enabling this config dynamically,
it needs to wait for the unclean leader election" +
+ "loss. Enabling this configuration for a topic that uses transactions
is incompatible with exactly-once " +
+ "semantics. An unclean election can remove a transaction's COMMIT or
ABORT marker from the elected replica, " +
+ "causing consumers with <code>isolation.level=read_committed</code> to
stop at the last stable offset. " +
+ "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. " +
+ "Verify the transaction and partition before aborting it, since an
unclean election may already have caused " +
+ "data loss." +
Review Comment:
```suggestion
```
##########
clients/src/main/java/org/apache/kafka/common/config/TopicConfig.java:
##########
@@ -199,7 +199,16 @@ public class TopicConfig {
public static final String UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG =
"unclean.leader.election.enable";
public static final String UNCLEAN_LEADER_ELECTION_ENABLE_DOC = "Indicates
whether to enable replicas " +
"not in the ISR set to be elected as leader as a last resort, even
though doing so may result in data " +
- "loss.<p>Note: In KRaft mode, when enabling this config dynamically,
it needs to wait for the unclean leader election" +
+ "loss. Enabling this configuration for a topic that uses transactions
is incompatible with exactly-once " +
+ "semantics. An unclean election can remove a transaction's COMMIT or
ABORT marker from the elected replica, " +
+ "causing consumers with <code>isolation.level=read_committed</code> to
stop at the last stable offset. " +
+ "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. " +
Review Comment:
```suggestion
"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. "
```
##########
server/src/test/java/org/apache/kafka/server/UncleanLeaderElectionTest.java:
##########
@@ -139,6 +152,126 @@ public void testUncleanLeaderElectionEnabledConsumer()
throws Exception {
testUncleanLeaderElectionEnabled(GroupProtocol.CONSUMER);
}
+ @ClusterTest(
+ serverProperties = {
+ @ClusterConfigProperty(key =
TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key =
TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key =
TransactionLogConfig.TRANSACTIONS_TOPIC_MIN_ISR_CONFIG, value = "1")
+ }
+ )
+ public void
testUncleanLeaderElectionCanLeaveReadCommittedConsumerAtLastStableOffset()
throws Exception {
+ disableEligibleLeaderReplicas();
+
+ String transactionalId = "unclean-election-" + UUID.randomUUID();
+ Map<String, Object> producerConfig = Map.of(
+ ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
cluster.bootstrapServers(),
+ ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName(),
+ ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName(),
+ ProducerConfig.ACKS_CONFIG, "all",
+ ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId
+ );
+
+ try (KafkaProducer<String, String> producer = new
KafkaProducer<>(producerConfig)) {
+ producer.initTransactions();
+
+ var transactionStateTopic =
admin.describeTopics(Set.of(Topic.TRANSACTION_STATE_TOPIC_NAME))
+
.allTopicNames().get().get(Topic.TRANSACTION_STATE_TOPIC_NAME);
+ int transactionPartitionId = Utils.abs(transactionalId.hashCode())
% transactionStateTopic.partitions().size();
+ int transactionCoordinatorId =
transactionStateTopic.partitions().stream()
+ .filter(partition -> partition.partition() ==
transactionPartitionId)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("Transaction state
partition is missing"))
+ .leader().id();
+ int followerId = transactionCoordinatorId == BROKER_ID_0 ?
BROKER_ID_1 : BROKER_ID_0;
+
+ NewTopic newTopic = new NewTopic(TOPIC, Map.of(PARTITION_ID,
List.of(transactionCoordinatorId, followerId)))
+ .configs(Map.of(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG,
"1"));
+ admin.createTopics(List.of(newTopic)).all().get();
+ try {
+ admin.electLeaders(ElectionType.PREFERRED,
Set.of(TOPIC_PARTITION)).all().get();
+ } catch (ExecutionException e) {
+ assertInstanceOf(ElectionNotNeededException.class,
e.getCause());
+ }
+
+ int leaderId = awaitLeaderChange(cluster, TOPIC_PARTITION,
Optional.of(transactionCoordinatorId));
+ KafkaBroker leader = cluster.brokers().get(leaderId);
+ KafkaBroker follower = cluster.brokers().get(followerId);
+
+ produceMessage(cluster, TOPIC, "before");
+ waitForCondition(
+ () -> {
+ var log =
follower.replicaManager().localLog(TOPIC_PARTITION);
+ return log.isDefined() && log.get().logEndOffset() >=
1;
+ },
+ DEFAULT_MAX_WAIT_MS,
+ "Timed out waiting for the follower to replicate the
non-transactional record"
+ );
+
+ producer.beginTransaction();
+ producer.send(new ProducerRecord<>(TOPIC, "transactional")).get();
+
+ // The follower must contain the transaction data, but must be
stopped before the marker is committed.
+ waitForCondition(
+ () -> {
+ var log =
follower.replicaManager().localLog(TOPIC_PARTITION);
+ return log.isDefined() && log.get().logEndOffset() >=
2;
+ },
+ DEFAULT_MAX_WAIT_MS,
+ "Timed out waiting for the follower to replicate the
transactional record"
+ );
+ follower.shutdown();
+ follower.awaitShutdown();
+ waitForCondition(
+ () -> {
+ var leaderAndIsr =
leader.replicaManager().metadataCache().getLeaderAndIsr(TOPIC, PARTITION_ID);
+ return leaderAndIsr.isPresent() &&
leaderAndIsr.get().isr().equals(Set.of(leaderId));
+ },
+ DEFAULT_MAX_WAIT_MS,
+ "Timed out waiting for the follower to leave the ISR"
+ );
+
+ // The marker is now written only to the leader. The follower will
retain the open transaction after failover.
+ producer.commitTransaction();
+
+ // Close the producer while its coordinator is still available; no
client cleanup should depend on the old leader.
+ producer.close();
+ leader.shutdown();
+ leader.awaitShutdown();
+ follower.startup();
+ awaitLeaderChange(cluster, TOPIC_PARTITION,
Optional.of(followerId));
+
+ 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());
+ }
+ }
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
