KAFKA-17407 Fix flaky ShareGroupHeartbeatRequestTest.testPartitionAssignmentWithChangingTopics (apache#16994)

Reviewers: Andrew Schofield <[email protected]>, Chia-Ping Tsai <[email protected]>

Author: m1a2st

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(