lucasbru commented on code in PR #23670:
URL: https://github.com/apache/kafka/pull/23670#discussion_r4165446247
##########
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:
Fair catch - raised group.streams.session.timeout.ms (and its max) to 120s
for this test, well above the 40s sleep.
--
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]
