andyhuangdev commented on code in PR #23670:
URL: https://github.com/apache/kafka/pull/23670#discussion_r4162070343
##########
core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionRequestTest.scala:
##########
@@ -321,6 +321,258 @@ 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()
+ }
+ }
+
+ @ClusterTest(serverProperties = Array(
+ new ClusterConfigProperty(
+ key =
GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG,
+ value = "kafka.server.FailingTopologyDescriptionPlugin")
+ ))
+ def testStreamsGroupTopologyDescriptionUpdateTransientFailureArmsBackOff():
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.TRANSIENT)
+
+ val updateResponse = streamsGroupTopologyDescriptionUpdate(
+ groupId = groupId,
+ memberId = memberId,
+ topologyEpoch = topologyEpoch,
+ topologyDescription = createTopologyDescription(topicName)
+ )
+ assertEquals(Errors.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED.code(),
updateResponse.errorCode())
+ assertEquals("backend offline", updateResponse.errorMessage())
+
+ // A transient failure arms the per-group back-off: a heartbeat
immediately afterwards
+ // must not re-solicit another push.
+ 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 immediately
after a transient failure; back-off must be armed.")
Review Comment:
Let the initial solicitation window expire before triggering the transient
failure (or control the broker clock/back-off), then verify suppression after
the failure and eventual re-solicitation at the same topology epoch.
--
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]
