This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 25c099e351c KAFKA-20806: Align failed-push expiration test with
UNCERTAIN reclaim (#22840)
25c099e351c is described below
commit 25c099e351c23acd8d8fed526dc43c93d1c15423
Author: TengYao Chi <[email protected]>
AuthorDate: Wed Jul 15 20:23:37 2026 +0100
KAFKA-20806: Align failed-push expiration test with UNCERTAIN reclaim
(#22840)
The test shouldExpireGroupWithoutDeleteTopologyWhenPushPermanentlyFailed
started failing because of commit f8c9566af1 (KAFKA-20748), which
deliberately changed the post-failure state of a Streams group.
Before KAFKA-20748:
- A permanent `setTopology` failure left
`StoredDescriptionTopologyEpoch` at `-1` (NONE).
- The cleanup scan filtered out groups with `stored == NONE`, so the
group was tombstoned directly without any `plugin.deleteTopology` call.
- The test's assertion `assertEquals(0, deleteTopologyCalls(appId))`
matched that invariant.
After KAFKA-20748:
1. The push path now writes an `UNCERTAIN(-2)` barrier to
`StoredDescriptionTopologyEpoch` before calling `plugin.setTopology` —
this is the durable marker that says "the plugin call is in progress; a
residual entry may exist" (GroupCoordinatorService.java:748).
2. When the plugin returns a permanent failure,
`setFailedDescriptionTopologyEpoch` only advances
`FailedDescriptionTopologyEpoch`; it deliberately leaves stored
untouched (GroupMetadataManager.java:8600-8612). So the group ends up
parked at `UNCERTAIN(-2)` rather than `NONE(-1)`.
3. The eligibility scan (GroupCoordinatorShard.java:1013-1032) now skips
only groups whose `storedEpoch == NONE`, so `UNCERTAIN` groups are
picked up.
4. The cleanup cycle therefore issues a defensive
`plugin.deleteTopology` on the group to reclaim any potentially leaked
plugin entry, then tombstones the group.
The single defensive `deleteTopology` call is the observed expected: <0>
but was: <1>. This is the intended new behavior — the plugin call and
the metadata write are non-atomic, so the broker cannot rule out a
residual entry after a permanent failure and must reclaim it
defensively.
Reviewers: Alieh Saeedi <[email protected]>, Matthias J. Sax
<[email protected]>
---
...DescriptionPluginExpirationIntegrationTest.java | 22 ++++++++++++++--------
1 file changed, 14 insertions(+), 8 deletions(-)
diff --git
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TopologyDescriptionPluginExpirationIntegrationTest.java
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TopologyDescriptionPluginExpirationIntegrationTest.java
index 06762f1c3fa..5a6dfd67fd4 100644
---
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TopologyDescriptionPluginExpirationIntegrationTest.java
+++
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/TopologyDescriptionPluginExpirationIntegrationTest.java
@@ -49,14 +49,15 @@ import java.util.Properties;
import java.util.concurrent.ExecutionException;
import static
org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
-import static org.junit.jupiter.api.Assertions.assertEquals;
/**
* Tests of the topology description plugin behavior when a streams group
expires naturally
* (KIP-1331): the periodic broker-side cleanup cycle calls {@code
plugin.deleteTopology} and
* clears the stored topology epoch for empty groups without offsets, after
which the regular
- * expiration sweep tombstones the group. Groups whose push permanently failed
have no stored
- * plugin state, so they are tombstoned directly without a {@code
deleteTopology} call.
+ * expiration sweep tombstones the group. Groups whose push permanently failed
are left at
+ * UNCERTAIN(-2) — the non-atomic plugin call and epoch write mean the broker
cannot rule out a
+ * residual plugin entry — so the cleanup cycle still issues {@code
plugin.deleteTopology}
+ * defensively before the tombstone.
*
* <p>The tests avoid producing any input data, so the groups never commit
offsets and become
* eligible for expiration as soon as their last member leaves.
@@ -119,7 +120,7 @@ public class
TopologyDescriptionPluginExpirationIntegrationTest {
}
@Test
- public void
shouldExpireGroupWithoutDeleteTopologyWhenPushPermanentlyFailed() throws
Exception {
+ public void shouldReclaimUncertainPluginStateWhenPushPermanentlyFailed()
throws Exception {
final String appId = "topology-description-failed-expiration-app";
final String inputTopic =
"topology-description-failed-expiration-input";
cluster.createTopic(inputTopic, 1, 1);
@@ -133,11 +134,16 @@ public class
TopologyDescriptionPluginExpirationIntegrationTest {
"Expected a setTopology call for group " + appId);
}
- // The push permanently failed, so no topology description is
stored for the
- // group: the tombstone fires directly, without any deleteTopology
call.
+ // The permanent push failure leaves the group at UNCERTAIN(-2):
the barrier written
+ // before the plugin call remains, because the failed-epoch write
only advances the
+ // failed epoch. UNCERTAIN is delete-eligible, so the cleanup
cycle must call
+ // plugin.deleteTopology defensively to reclaim any residual entry
before the
+ // expiration sweep tombstones the group.
+ TestUtils.waitForCondition(
+ () ->
TrackingTopologyDescriptionPlugin.deleteTopologyCalls(appId) >= 1,
+ "Expected the cleanup cycle to invoke deleteTopology for group
" + appId
+ + " left at UNCERTAIN by the permanent push failure");
waitForGroupToBeDeleted(admin, appId);
- assertEquals(0,
TrackingTopologyDescriptionPlugin.deleteTopologyCalls(appId),
- "Expected no deleteTopology call for group " + appId + " whose
push permanently failed");
}
}
