Copilot commented on code in PR #23670:
URL: https://github.com/apache/kafka/pull/23670#discussion_r4164795874
##########
core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionRequestTest.scala:
##########
@@ -321,6 +323,282 @@ class StreamsGroupTopologyDescriptionRequestTest(cluster:
ClusterInstance) exten
}
}
+ @ClusterTest(serverProperties = Array(
+ new ClusterConfigProperty(
+ key =
GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG,
+ value = "kafka.server.FailingTopologyDescriptionPlugin")
+ ))
+ def
testStreamsGroupTopologyDescriptionUpdatePermanentFailureRatchetsFailedEpoch():
Unit = {
+ val admin = cluster.admin()
+ val groupId = "test-group"
+ val memberId = "test-member"
+ val topicName = "test-topic"
+
+ try {
+ FailingTopologyDescriptionPlugin.reset()
+ TestUtils.createOffsetsTopicWithAdmin(
+ admin = admin,
+ brokers = cluster.brokers.values().asScala.toSeq,
+ controllers = cluster.controllers().values().asScala.toSeq
+ )
+ TestUtils.createTopicWithAdmin(
+ admin = admin,
+ brokers = cluster.brokers.values().asScala.toSeq,
+ controllers = cluster.controllers().values().asScala.toSeq,
+ topic = topicName,
+ numPartitions = 3
+ )
+
+ // Join the group and wait until the broker solicits a topology
description push.
+ var memberEpoch = 0
+ TestUtils.waitUntilTrue(() => {
+ val response = streamsGroupHeartbeat(
+ groupId = groupId,
+ memberId = memberId,
+ rebalanceTimeoutMs = 1000,
+ activeTasks = List.empty,
+ standbyTasks = List.empty,
+ warmupTasks = List.empty,
+ topology = createMockTopology(topicName)
+ )
+ memberEpoch = response.memberEpoch()
+ response.errorCode == Errors.NONE.code() &&
response.topologyDescriptionRequired()
+ }, "Broker did not solicit a topology description push within the
timeout period.")
+
+
FailingTopologyDescriptionPlugin.failNextSetTopology(FailingTopologyDescriptionPlugin.SetTopologyFailureMode.PERMANENT)
+
+ val updateResponse = streamsGroupTopologyDescriptionUpdate(
+ groupId = groupId,
+ memberId = memberId,
+ topologyEpoch = topologyEpoch,
+ topologyDescription = createTopologyDescription(topicName)
+ )
+ assertEquals(Errors.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED.code(),
updateResponse.errorCode())
+ assertEquals("topology rejected by test plugin",
updateResponse.errorMessage())
+
+ // A permanent failure ratchets the group's failed topology epoch: the
broker must not
+ // re-solicit another push at the same epoch.
+ val heartbeatAfterFailure = streamsGroupHeartbeat(
+ groupId = groupId,
+ memberId = memberId,
+ memberEpoch = memberEpoch,
+ rebalanceTimeoutMs = 1000,
+ activeTasks = List.empty,
+ standbyTasks = List.empty,
+ warmupTasks = List.empty
+ )
+ assertFalse(heartbeatAfterFailure.topologyDescriptionRequired(),
+ "Broker must not re-solicit a topology description push once the epoch
is marked permanently failed.")
+
+ // Nothing was ever stored for this epoch.
+ val describedGroup = streamsGroupDescribe(
+ groupIds = List(groupId),
+ includeTopologyDescription = true
+ ).head
+
assertNotEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_AVAILABLE,
describedGroup.topologyDescriptionStatus())
+ } finally {
+ FailingTopologyDescriptionPlugin.reset()
+ admin.close()
+ }
+ }
+
+ @Timeout(180)
+ @ClusterTest(serverProperties = Array(
+ new ClusterConfigProperty(
+ key =
GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG,
+ value = "kafka.server.FailingTopologyDescriptionPlugin")
+ ))
Review Comment:
This test later stops heartbeating for 40 seconds, while the default Streams
session timeout is only 45 seconds (`GroupCoordinatorConfig.java:343-344`).
That leaves just five seconds for CI or scheduler delays before the member
expires, making the update/heartbeat fail for membership reasons instead of
testing transient back-off. Configure a session timeout comfortably above the
sleep (and raise its maximum accordingly) for this test.
--
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]
