(kafka) branch trunk updated: KAFKA-17407 Fix flaky ShareGroupHeartbeatRequestTest.testPartitionAssignmentWithChangingTopics (#16994)

[ch...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 2267518d5be KAFKA-17407 Fix flaky ShareGroupHeartbeatRequestTest.testPartitionAssignmentWithChangingTopics (#16994)
2267518d5be is described below

commit 2267518d5bed60e2a1731712ae7cb455ea0bf8c4
Author: Ken Huang <s7...@gmail.com>
AuthorDate: Sat Oct 19 18:45:04 2024 +0800

    KAFKA-17407 Fix flaky ShareGroupHeartbeatRequestTest.testPartitionAssignmentWithChangingTopics (#16994)
    
    Reviewers: Andrew Schofield <as...@confluent.io>, Chia-Ping Tsai <ch...@gmail.com>
---
 .../kafka/server/ShareGroupHeartbeatRequestTest.scala | 19 +++++++++++--------
 1 file changed, 11 insertions(+), 8 deletions(-)

diff --git a/core/src/test/scala/unit/kafka/server/ShareGroupHeartbeatRequestTest.scala b/core/src/test/scala/unit/kafka/server/ShareGroupHeartbeatRequestTest.scala
index 47287f03ae4..fa9ad9da7b2 100644
--- a/core/src/test/scala/unit/kafka/server/ShareGroupHeartbeatRequestTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ShareGroupHeartbeatRequestTest.scala
@@ -407,7 +407,7 @@ class ShareGroupHeartbeatRequestTest(cluster: ClusterInstance) {
       controllers = raftCluster.controllers().values().asScala.toSeq
     )
     // Heartbeat request to join the group. Note that the member subscribes
-    // to an nonexistent topic.
+    // to a nonexistent topic.
     var shareGroupHeartbeatRequest = new ShareGroupHeartbeatRequest.Builder(
       new ShareGroupHeartbeatRequestData()
         .setGroupId("grp")
@@ -457,6 +457,9 @@ class ShareGroupHeartbeatRequestTest(cluster: ClusterInstance) {
       true
     ).build()
 
+    cluster.waitForTopic("foo", 2)
+    cluster.waitForTopic("bar", 3)
+
     TestUtils.waitUntilTrue(() => {
       shareGroupHeartbeatResponse = connectAndReceive(shareGroupHeartbeatRequest)
       shareGroupHeartbeatResponse.data.errorCode == Errors.NONE.code &&
@@ -465,7 +468,7 @@ class ShareGroupHeartbeatRequestTest(cluster: ClusterInstance) {
         shareGroupHeartbeatResponse.data.assignment.topicPartitions.containsAll(expectedAssignment.topicPartitions)
     }, msg = s"Could not get partitions for topic foo and bar assigned. Last response $shareGroupHeartbeatResponse.")
     // Verify the response.
-    assertEquals(3, shareGroupHeartbeatResponse.data.memberEpoch)
+    assertEquals(2, shareGroupHeartbeatResponse.data.memberEpoch)
     // Create the topic baz.
     val bazTopicId = TestUtils.createTopicWithAdminRaw(
       admin = admin,
@@ -489,7 +492,7 @@ class ShareGroupHeartbeatRequestTest(cluster: ClusterInstance) {
       new ShareGroupHeartbeatRequestData()
         .setGroupId("grp")
         .setMemberId(memberId)
-        .setMemberEpoch(3),
+        .setMemberEpoch(2),
       true
     ).build()
 
@@ -501,7 +504,7 @@ class ShareGroupHeartbeatRequestTest(cluster: ClusterInstance) {
         shareGroupHeartbeatResponse.data.assignment.topicPartitions.containsAll(expectedAssignment.topicPartitions)
     }, msg = s"Could not get partitions for topic baz assigned. Last response $shareGroupHeartbeatResponse.")
     // Verify the response.
-    assertEquals(4, shareGroupHeartbeatResponse.data.memberEpoch)
+    assertEquals(3, shareGroupHeartbeatResponse.data.memberEpoch)
     // Increasing the partitions of topic bar which is already being consumed in the share group.
     increasePartitions(admin, "bar", 6, Seq.empty)
 
@@ -521,7 +524,7 @@ class ShareGroupHeartbeatRequestTest(cluster: ClusterInstance) {
       new ShareGroupHeartbeatRequestData()
         .setGroupId("grp")
         .setMemberId(memberId)
-        .setMemberEpoch(4),
+        .setMemberEpoch(3),
       true
     ).build()
 
@@ -533,7 +536,7 @@ class ShareGroupHeartbeatRequestTest(cluster: ClusterInstance) {
         shareGroupHeartbeatResponse.data.assignment.topicPartitions.containsAll(expectedAssignment.topicPartitions)
     }, msg = s"Could not update partitions assignment for topic bar. Last response $shareGroupHeartbeatResponse.")
     // Verify the response.
-    assertEquals(5, shareGroupHeartbeatResponse.data.memberEpoch)
+    assertEquals(4, shareGroupHeartbeatResponse.data.memberEpoch)
     // Delete the topic foo.
     TestUtils.deleteTopicWithAdmin(
       admin = admin,
@@ -555,7 +558,7 @@ class ShareGroupHeartbeatRequestTest(cluster: ClusterInstance) {
       new ShareGroupHeartbeatRequestData()
         .setGroupId("grp")
         .setMemberId(memberId)
-        .setMemberEpoch(5),
+        .setMemberEpoch(4),
       true
     ).build()
 
@@ -567,7 +570,7 @@ class ShareGroupHeartbeatRequestTest(cluster: ClusterInstance) {
         shareGroupHeartbeatResponse.data.assignment.topicPartitions.containsAll(expectedAssignment.topicPartitions)
     }, msg = s"Could not update partitions assignment for topic foo. Last response $shareGroupHeartbeatResponse.")
     // Verify the response.
-    assertEquals(6, shareGroupHeartbeatResponse.data.memberEpoch)
+    assertEquals(5, shareGroupHeartbeatResponse.data.memberEpoch)
   }
 
   @ClusterTest(