Shekharrajak commented on code in PR #22357:
URL: https://github.com/apache/kafka/pull/22357#discussion_r3430597871
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTransactionTest.java:
##########
@@ -501,4 +571,27 @@ private int shareStatePartition(String groupId, Uuid
topicId, int partition) {
SharePartitionKey key = SharePartitionKey.getInstance(groupId,
topicId, partition);
return Utils.abs(key.asCoordinatorKey().hashCode()) %
sgsTopicPartitions.size();
}
+
+ private void shutdownLeaderAndWaitForNewLeader(Admin admin, TopicPartition
topicPartition) throws Exception {
+ List<Integer> oldLeaders = topicPartitionLeader(admin,
topicPartition.topic(), topicPartition.partition());
+ assertEquals(1, oldLeaders.size());
+ int oldLeader = oldLeaders.get(0);
+ KafkaBroker broker = cluster.brokers().get(oldLeader);
+ cluster.shutdownBroker(oldLeader);
+ broker.awaitShutdown();
+ TestUtils.waitForCondition(
Review Comment:
new topic partition leader - simulating broker failure and leader is changed
after txn_pending state change and before txn commit happens
--
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]
