lucasbru commented on code in PR #23670:
URL: https://github.com/apache/kafka/pull/23670#discussion_r4209763835
##########
core/src/test/scala/unit/kafka/server/StreamsGroupTopologyDescriptionRequestTest.scala:
##########
@@ -321,6 +322,280 @@ 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
+ )
+
+ val memberEpoch = joinAndAwaitTopologyDescriptionSolicited(groupId,
memberId, topicName)
+
+
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: the permanent failure leaves
the group's
+ // stored-topology-epoch at its UNCERTAIN sentinel, which Describe
reports as NOT_STORED.
+ val describedGroup = streamsGroupDescribe(
+ groupIds = List(groupId),
+ includeTopologyDescription = true
+ ).head
+
assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED,
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"),
+ // The test below goes 40s without heartbeating; the default 45s session
timeout would
+ // leave only ~5s of margin before the member is reaped for inactivity.
+ new ClusterConfigProperty(key =
GroupCoordinatorConfig.STREAMS_GROUP_SESSION_TIMEOUT_MS_CONFIG, value =
"120000"),
+ new ClusterConfigProperty(key =
GroupCoordinatorConfig.STREAMS_GROUP_MAX_SESSION_TIMEOUT_MS_CONFIG, value =
"120000")
+ ))
+ 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. This
+ // arms the first back-off window: 30s * 2^0, +/-20% jitter
(StreamsGroupTopologyDescriptionBackoff),
+ // i.e. 24-36s.
+ var memberEpoch = joinAndAwaitTopologyDescriptionSolicited(groupId,
memberId, topicName)
+
+ // Let that window fully expire without heartbeating, so nothing but the
upcoming
+ // transient failure can be responsible for the suppression asserted
below. The back-off
+ // uses the broker's real clock, with no test seam to inject a mock one
here, so this
+ // waits out real time rather than simulating it.
+ Thread.sleep(40000)
Review Comment:
Couldn't find a good way to avoid the sleep here.
--
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]
