This is an automated email from the ASF dual-hosted git repository.
lucasbru 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 4e18e9b1633 KAFKA-20623: Heartbeat extension for streams group
topology description plugin (1/3) (#22551)
4e18e9b1633 is described below
commit 4e18e9b1633fe98cf4dd342fab23b59885327f64
Author: TengYao Chi <[email protected]>
AuthorDate: Tue Jun 16 08:14:43 2026 +0100
KAFKA-20623: Heartbeat extension for streams group topology description
plugin (1/3) (#22551)
JIRA: KAFKA-20623 This PR is a part of KIP-1331
Wires the plugin reference into `GroupCoordinatorService` and adds the
heartbeat-path gate that asks streams clients to push their topology
description.
### Plugin reference on the service
- `GroupCoordinatorService.Builder.build()` resolves the plugin
internally via
`config.streamsGroupTopologyDescriptionPlugin(Map.of())`
- The service constructor accepts
`Optional<StreamsGroupTopologyDescriptionPlugin>` and
hands it to a new `TopologyDescriptionManager` that owns the plugin
reference and the
per-group push back-off.
### Heartbeat post-processing
- `StreamsGroupHeartbeatResult` carries three epoch fields now —
`currentTopologyEpoch`, `storedDescriptionTopologyEpoch`,
`failedDescriptionTopologyEpoch` — so the service-layer gate can
decide whether to set
`TopologyDescriptionRequired=true` without re-reading the group on
every heartbeat.
- `GroupMetadataManager` builds the heartbeat result with these fields
at the four
existing `StreamsGroupHeartbeatResult` construction sites.
- `TopologyDescriptionManager.maybeSetTopologyDescriptionRequired(...)`
runs in the
`.thenApply(...)` after the heartbeat write. The flag is set when the
plugin is
configured, the response has no error, the current epoch is resolved,
that epoch is
neither stored nor permanently failed at the plugin, the response does
not carry a
`STALE_TOPOLOGY` status, and the per-group back-off window is not in
effect.
### Back-off
- `StreamsGroupTopologyDescriptionBackoff` is a broker-level, per-group
exponential
back-off (30 s → 1 h, doubled on each arm at the same topology epoch,
reset on
topology-epoch advance).
- The heartbeat path uses a single atomic `armIfNotActive` compute that
serves two
purposes: it prevents two concurrent heartbeats for the same group
from both
arming the back-off, and — when the previous window expired without a
push
reaching the coordinator (client opted out via
`topology.description.push.enabled=false`,
client never recorded the flag, or the push was lost in flight) — it
continues
the exponential chain across the re-arm instead of resetting to the
initial
delay. Without this, the heartbeat-only re-solicit loop would
degenerate into a
fixed 30 s cadence and the documented exponential throttle would never
engage.
- The back-off is non-timeline, non-replayed state — rebuilt from
scratch on broker
restart. Convergence after a restart is driven by the persisted
`StoredDescriptionTopologyEpoch` / `FailedDescriptionTopologyEpoch`
fields on each
streams group.
- The `clear(...)` and `armOrExtend(...)` sites consumed by the push and
DeleteGroups
paths land in the follow-up PRs along with their respective entry
points on
`TopologyDescriptionManager`.
Reviewers: Lucas Brutschy <[email protected]>
---
.../scala/unit/kafka/server/KafkaApisTest.scala | 10 +-
.../coordinator/group/GroupCoordinatorService.java | 39 ++-
.../coordinator/group/GroupMetadataManager.java | 32 ++-
.../group/streams/StreamsGroupHeartbeatResult.java | 32 ++-
.../StreamsGroupTopologyDescriptionBackoff.java | 114 +++++++++
.../StreamsGroupTopologyDescriptionManager.java | 129 ++++++++++
.../group/GroupCoordinatorServiceTest.java | 53 +++-
...pCoordinatorServiceTopologyDescriptionTest.java | 285 +++++++++++++++++++++
.../group/GroupCoordinatorShardTest.java | 2 +-
.../streams/StreamsGroupHeartbeatResultTest.java | 31 ++-
...StreamsGroupTopologyDescriptionBackoffTest.java | 152 +++++++++++
11 files changed, 839 insertions(+), 40 deletions(-)
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index e9542974410..701f2db0df0 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
@@ -11049,7 +11049,7 @@ class KafkaApisTest extends Logging {
val streamsGroupHeartbeatResponse = new StreamsGroupHeartbeatResponseData()
.setMemberId("member")
- future.complete(new
StreamsGroupHeartbeatResult(streamsGroupHeartbeatResponse, util.Map.of(), -1))
+ future.complete(new
StreamsGroupHeartbeatResult(streamsGroupHeartbeatResponse, util.Map.of(), -1,
-1, -1))
val response =
verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
assertEquals(streamsGroupHeartbeatResponse, response.data)
}
@@ -11119,7 +11119,7 @@ class KafkaApisTest extends Logging {
val streamsGroupHeartbeatResponse = new StreamsGroupHeartbeatResponseData()
.setMemberId("member")
- future.complete(new
StreamsGroupHeartbeatResult(streamsGroupHeartbeatResponse, util.Map.of(), -1))
+ future.complete(new
StreamsGroupHeartbeatResult(streamsGroupHeartbeatResponse, util.Map.of(), -1,
-1, -1))
val response =
verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
assertEquals(streamsGroupHeartbeatResponse, response.data)
}
@@ -11351,7 +11351,7 @@ class KafkaApisTest extends Logging {
val streamsGroupHeartbeatResponse = new StreamsGroupHeartbeatResponseData()
.setMemberId("member")
- future.complete(new
StreamsGroupHeartbeatResult(streamsGroupHeartbeatResponse, missingTopics, -1))
+ future.complete(new
StreamsGroupHeartbeatResult(streamsGroupHeartbeatResponse, missingTopics, -1,
-1, -1))
val response =
verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
assertEquals(streamsGroupHeartbeatResponse, response.data)
verify(autoTopicCreationManager).createStreamsInternalTopics(any(), any(),
anyLong())
@@ -11400,7 +11400,7 @@ class KafkaApisTest extends Logging {
val streamsGroupHeartbeatResponse = new StreamsGroupHeartbeatResponseData()
.setMemberId("member")
- future.complete(new
StreamsGroupHeartbeatResult(streamsGroupHeartbeatResponse, missingTopics, -1))
+ future.complete(new
StreamsGroupHeartbeatResult(streamsGroupHeartbeatResponse, missingTopics, -1,
-1, -1))
val response =
verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
assertEquals(Errors.NONE.code, response.data.errorCode())
assertEquals(null, response.data.errorMessage())
@@ -11451,7 +11451,7 @@ class KafkaApisTest extends Logging {
.setStatusDetail("Internal topics are missing: [test-topic]")
))
- future.complete(new
StreamsGroupHeartbeatResult(streamsGroupHeartbeatResponse, missingTopics, -1))
+ future.complete(new
StreamsGroupHeartbeatResult(streamsGroupHeartbeatResponse, missingTopics, -1,
-1, -1))
val response =
verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
assertEquals(Errors.NONE.code, response.data.errorCode())
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
index e879c7083fe..303a7d1f2e0 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
@@ -98,9 +98,11 @@ import
org.apache.kafka.coordinator.common.runtime.MultiThreadedEventProcessor;
import org.apache.kafka.coordinator.common.runtime.PartitionWriter;
import org.apache.kafka.coordinator.group.GroupCoordinatorShard.DeletedTopic;
import
org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
+import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
import org.apache.kafka.coordinator.group.streams.StreamsGroupDescribeResult;
import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
+import
org.apache.kafka.coordinator.group.streams.StreamsGroupTopologyDescriptionManager;
import org.apache.kafka.image.MetadataDelta;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.image.TopicsDelta;
@@ -250,6 +252,9 @@ public class GroupCoordinatorService implements
GroupCoordinator {
String logPrefix = String.format("GroupCoordinator id=%d", nodeId);
LogContext logContext = new LogContext(String.format("[%s] ",
logPrefix));
+ Optional<StreamsGroupTopologyDescriptionPlugin>
streamsGroupTopologyDescriptionPlugin =
+
Optional.ofNullable(config.streamsGroupTopologyDescriptionPlugin(Map.of()));
+
CoordinatorShardBuilderSupplier<GroupCoordinatorShard,
CoordinatorRecord> supplier = () ->
new GroupCoordinatorShard.Builder(config, groupConfigManager)
.withAuthorizerPlugin(authorizerPlugin);
@@ -297,7 +302,9 @@ public class GroupCoordinatorService implements
GroupCoordinator {
groupConfigManager,
persister,
timer,
- partitionMetadataClient
+ partitionMetadataClient,
+ streamsGroupTopologyDescriptionPlugin,
+ time
);
}
}
@@ -352,6 +359,14 @@ public class GroupCoordinatorService implements
GroupCoordinator {
*/
private final PartitionMetadataClient partitionMetadataClient;
+ /**
+ * The broker-level component that owns the streams-group topology
description plugin
+ * (KIP-1331): plugin reference, per-group push back-off, and the three
entry points
+ * the service delegates into — heartbeat post-processing, the push RPC,
and the
+ * pre-tombstone hook on DeleteGroups.
+ */
+ private final StreamsGroupTopologyDescriptionManager
streamsGroupTopologyDescriptionManager;
+
/**
* The number of partitions of the __consumer_offsets topics. This is
provided
* when the component is started.
@@ -382,7 +397,9 @@ public class GroupCoordinatorService implements
GroupCoordinator {
GroupConfigManager groupConfigManager,
Persister persister,
Timer timer,
- PartitionMetadataClient partitionMetadataClient
+ PartitionMetadataClient partitionMetadataClient,
+ Optional<StreamsGroupTopologyDescriptionPlugin>
streamsGroupTopologyDescriptionPlugin,
+ Time time
) {
this.log = logContext.logger(GroupCoordinatorService.class);
this.config = config;
@@ -397,6 +414,10 @@ public class GroupCoordinatorService implements
GroupCoordinator {
.map(ConsumerGroupPartitionAssignor::name)
.collect(Collectors.toSet());
this.partitionMetadataClient = partitionMetadataClient;
+ this.streamsGroupTopologyDescriptionManager = new
StreamsGroupTopologyDescriptionManager(
+ streamsGroupTopologyDescriptionPlugin,
+ time
+ );
}
/**
@@ -606,9 +627,7 @@ public class GroupCoordinatorService implements
GroupCoordinator {
if (!isActive.get()) {
return CompletableFuture.completedFuture(
new StreamsGroupHeartbeatResult(
- new
StreamsGroupHeartbeatResponseData().setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code()),
- Map.of(),
- -1
+ new
StreamsGroupHeartbeatResponseData().setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
)
);
}
@@ -622,9 +641,7 @@ public class GroupCoordinatorService implements
GroupCoordinator {
new StreamsGroupHeartbeatResult(
new StreamsGroupHeartbeatResponseData()
.setErrorCode(apiError.error().code())
- .setErrorMessage(apiError.message()),
- Map.of(),
- -1
+ .setErrorMessage(apiError.message())
)
);
}
@@ -633,6 +650,7 @@ public class GroupCoordinatorService implements
GroupCoordinator {
"streams-group-heartbeat",
topicPartitionFor(request.groupId()),
coordinator -> coordinator.streamsGroupHeartbeat(context, request)
+ ).thenApply(result ->
streamsGroupTopologyDescriptionManager.maybeSetTopologyDescriptionRequired(result,
request.groupId(), context.requestVersion())
).exceptionally(exception -> handleOperationException(
"streams-group-heartbeat",
request,
@@ -641,9 +659,7 @@ public class GroupCoordinatorService implements
GroupCoordinator {
new StreamsGroupHeartbeatResult(
new StreamsGroupHeartbeatResponseData()
.setErrorCode(error.code())
- .setErrorMessage(message),
- Map.of(),
- -1
+ .setErrorMessage(message)
),
log
));
@@ -2380,6 +2396,7 @@ public class GroupCoordinatorService implements
GroupCoordinator {
log.info("Shutting down.");
isActive.set(false);
Utils.closeQuietly(runtime, "coordinator runtime");
+ Utils.closeQuietly(streamsGroupTopologyDescriptionManager, "streams
group topology description manager");
Utils.closeQuietly(groupCoordinatorMetrics, "group coordinator
metrics");
Utils.closeQuietly(groupConfigManager, "group config manager");
log.info("Shutdown complete.");
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
index 1f09b2908d9..c5673613fe4 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
@@ -2325,7 +2325,13 @@ public class GroupMetadataManager {
response.setStatus(returnedStatus);
- return new CoordinatorResult<>(records, new
StreamsGroupHeartbeatResult(response, internalTopicsToBeCreated,
updatedTopology.topologyEpoch()));
+ return new CoordinatorResult<>(records, new
StreamsGroupHeartbeatResult(
+ response,
+ internalTopicsToBeCreated,
+ updatedTopology.topologyEpoch(),
+ group.storedDescriptionTopologyEpoch(),
+ group.failedDescriptionTopologyEpoch()
+ ));
}
/**
@@ -4466,7 +4472,13 @@ public class GroupMetadataManager {
if (instanceId == null) {
StreamsGroupMember member = group.getMemberOrThrow(memberId);
log.info("[GroupId {}][MemberId {}] Member {} left the streams
group.", groupId, memberId, memberId);
- return streamsGroupFenceMember(group, member, new
StreamsGroupHeartbeatResult(response, Map.of(), group.currentTopologyEpoch()));
+ return streamsGroupFenceMember(group, member, new
StreamsGroupHeartbeatResult(
+ response,
+ Map.of(),
+ group.currentTopologyEpoch(),
+ group.storedDescriptionTopologyEpoch(),
+ group.failedDescriptionTopologyEpoch()
+ ));
} else {
StreamsGroupMember member = group.staticMember(instanceId);
throwIfStaticMemberIsUnknown(member, instanceId);
@@ -4478,7 +4490,13 @@ public class GroupMetadataManager {
} else {
log.info("[GroupId {}][MemberId {}] Static member {} with
instance id {} left the streams group.",
group.groupId(), memberId, memberId, instanceId);
- return streamsGroupFenceMember(group, member, new
StreamsGroupHeartbeatResult(response, Map.of(), group.currentTopologyEpoch()));
+ return streamsGroupFenceMember(group, member, new
StreamsGroupHeartbeatResult(
+ response,
+ Map.of(),
+ group.currentTopologyEpoch(),
+ group.storedDescriptionTopologyEpoch(),
+ group.failedDescriptionTopologyEpoch()
+ ));
}
}
}
@@ -4547,7 +4565,13 @@ public class GroupMetadataManager {
return new CoordinatorResult<>(
List.of(record),
- new StreamsGroupHeartbeatResult(response, Map.of(),
group.currentTopologyEpoch())
+ new StreamsGroupHeartbeatResult(
+ response,
+ Map.of(),
+ group.currentTopologyEpoch(),
+ group.storedDescriptionTopologyEpoch(),
+ group.failedDescriptionTopologyEpoch()
+ )
);
}
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResult.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResult.java
index b6066b41168..82b84d7b494 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResult.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResult.java
@@ -26,20 +26,40 @@ import java.util.Objects;
/**
* A simple record to hold the result of a StreamsGroupHeartbeat request.
*
- * @param data The data to be returned to the client.
- * @param creatableTopics The internal topics to be created.
- * @param currentTopologyEpoch The topology epoch the group is operating at
after this heartbeat, or -1 if the
- * group has no topology yet. The service layer
uses this to decide whether to set
- * TopologyDescriptionRequired on the response
(KIP-1331).
+ * <p>The three epoch fields let the service layer decide, without re-reading
the group,
+ * whether to set {@code TopologyDescriptionRequired} on the response: a push
is needed
+ * when the stored epoch lags the current epoch and the same epoch has not
already been
+ * recorded as a permanent failure. All three are -1 for failure-fast paths
that do not
+ * resolve a group.
+ *
+ * @param data The data to be returned to the
client.
+ * @param creatableTopics The internal topics to be created.
+ * @param currentTopologyEpoch The topology epoch the group is
operating at after this heartbeat,
+ * or -1 if the group has no topology
yet.
+ * @param storedDescriptionTopologyEpoch The most recent topology epoch
successfully stored by the topology
+ * description plugin, or -1 if none.
+ * @param failedDescriptionTopologyEpoch The most recent topology epoch the
plugin permanently rejected,
+ * or -1 if none.
*/
public record StreamsGroupHeartbeatResult(
StreamsGroupHeartbeatResponseData data,
Map<String, CreatableTopic> creatableTopics,
- int currentTopologyEpoch
+ int currentTopologyEpoch,
+ int storedDescriptionTopologyEpoch,
+ int failedDescriptionTopologyEpoch
) {
public StreamsGroupHeartbeatResult {
Objects.requireNonNull(data);
creatableTopics =
Collections.unmodifiableMap(Objects.requireNonNull(creatableTopics));
}
+
+ /**
+ * Convenience constructor for failure-fast paths that do not resolve a
group: no
+ * internal topics to create, and all three epoch fields set to -1 so the
service-layer
+ * gate sees nothing to do.
+ */
+ public StreamsGroupHeartbeatResult(StreamsGroupHeartbeatResponseData data)
{
+ this(data, Map.of(), -1, -1, -1);
+ }
}
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionBackoff.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionBackoff.java
new file mode 100644
index 00000000000..afda7817908
--- /dev/null
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionBackoff.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams;
+
+import org.apache.kafka.common.utils.Time;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * In-memory per-group back-off that throttles broker re-solicitation of a
topology
+ * description push. An entry is armed when the broker decides to set
+ * {@code TopologyDescriptionRequired=true} on a heartbeat or after a
transient plugin
+ * failure; consecutive arms at the same topology epoch double the window from
+ * {@value #INITIAL_DELAY_MS} ms up to {@value #MAX_DELAY_MS} ms. Successful
pushes,
+ * permanent plugin failures, and topology-epoch advances clear the entry.
+ */
+public class StreamsGroupTopologyDescriptionBackoff {
+
+ static final long INITIAL_DELAY_MS = 30_000L;
+ static final long MAX_DELAY_MS = 3_600_000L;
+
+ private final Time time;
+ private final ConcurrentHashMap<String, Entry> state = new
ConcurrentHashMap<>();
+
+ record Entry(int topologyEpoch, long currentDelayMs, long nextAttemptMs) {
}
+
+ public StreamsGroupTopologyDescriptionBackoff(Time time) {
+ this.time = time;
+ }
+
+ /**
+ * @return true if a back-off window is in effect for the given group at
the given
+ * topology epoch and the broker should suppress soliciting
another push.
+ */
+ public boolean isActive(String groupId, int topologyEpoch) {
+ Entry entry = state.get(groupId);
+ return entry != null
+ && entry.topologyEpoch() == topologyEpoch
+ && time.milliseconds() < entry.nextAttemptMs();
+ }
+
+ /**
+ * Atomic check-and-arm. Returns true if no window was in effect and a new
one was
+ * armed, false if a window was already active and nothing changed. Used
on the
+ * heartbeat path to fold the "check + arm" pair into a single compute so
two
+ * concurrent heartbeats for the same group cannot both arm the back-off,
and to
+ * preserve the exponential chain when the previous window has expired
without a
+ * push reaching the coordinator (e.g. the client never sent one, or the
push was
+ * lost in flight before {@link #armOrExtend} could run).
+ */
+ public boolean armIfNotActive(String groupId, int topologyEpoch) {
+ final long now = time.milliseconds();
+ final boolean[] armed = new boolean[]{false};
+ state.compute(groupId, (key, existing) -> {
+ if (existing != null
+ && existing.topologyEpoch() == topologyEpoch
+ && now < existing.nextAttemptMs()) {
+ return existing;
+ }
+ armed[0] = true;
+ // Re-arm: continue the exponential chain at the same epoch (the
previous
+ // window expired without a push completing); reset to
INITIAL_DELAY_MS on
+ // an epoch advance, which implicitly drops the prior history.
+ if (existing != null && existing.topologyEpoch() == topologyEpoch)
{
+ long nextDelay = Math.min(existing.currentDelayMs() * 2,
MAX_DELAY_MS);
+ return new Entry(topologyEpoch, nextDelay, now + nextDelay);
+ }
+ return new Entry(topologyEpoch, INITIAL_DELAY_MS, now +
INITIAL_DELAY_MS);
+ });
+ return armed[0];
+ }
+
+ /**
+ * Arm a new back-off window or extend the existing one. If the existing
entry is for a
+ * different topology epoch the window is reset to {@link
#INITIAL_DELAY_MS}.
+ */
+ public void armOrExtend(String groupId, int topologyEpoch) {
+ final long now = time.milliseconds();
+ state.compute(groupId, (key, existing) -> {
+ if (existing == null || existing.topologyEpoch() != topologyEpoch)
{
+ return new Entry(topologyEpoch, INITIAL_DELAY_MS, now +
INITIAL_DELAY_MS);
+ }
+ long nextDelay = Math.min(existing.currentDelayMs() * 2,
MAX_DELAY_MS);
+ return new Entry(topologyEpoch, nextDelay, now + nextDelay);
+ });
+ }
+
+ /**
+ * Drop the back-off entry for a group. Called on a successful push, a
permanent plugin
+ * failure, or when the group is removed.
+ */
+ public void clear(String groupId) {
+ state.remove(groupId);
+ }
+
+ // Visible for testing.
+ Entry entry(String groupId) {
+ return state.get(groupId);
+ }
+}
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
new file mode 100644
index 00000000000..ca848518d10
--- /dev/null
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams;
+
+import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse.Status;
+import org.apache.kafka.common.utils.Time;
+import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin;
+
+import java.util.Optional;
+
+/**
+ * Broker-level component that owns everything tied to the streams-group
topology
+ * description plugin: the configured plugin reference, the per-group
re-solicitation
+ * back-off, and the heartbeat-path gate that asks clients to push their
topology.
+ *
+ * <p>This class is broker-level (one instance per {@code
GroupCoordinatorService}); the
+ * back-off map is keyed by {@code groupId} and shared across all partitions
hosted on the
+ * broker. State here is intentionally non-timeline and non-replayed: it is
rebuilt from
+ * scratch on broker restart, and the persisted {@code
StoredDescriptionTopologyEpoch} /
+ * {@code FailedDescriptionTopologyEpoch} fields on each streams group drive
convergence
+ * after a restart.
+ */
+public class StreamsGroupTopologyDescriptionManager implements AutoCloseable {
+ private final Optional<StreamsGroupTopologyDescriptionPlugin> plugin;
+ private final StreamsGroupTopologyDescriptionBackoff backoff;
+
+ public StreamsGroupTopologyDescriptionManager(
+ Optional<StreamsGroupTopologyDescriptionPlugin> plugin,
+ Time time
+ ) {
+ this.plugin = plugin;
+ this.backoff = new StreamsGroupTopologyDescriptionBackoff(time);
+ }
+
+ /**
+ * Release plugin-side resources. The plugin is instantiated by the
service via
+ * {@code config.getConfiguredInstance(...)}, so the service owns it and
must close
+ * it on shutdown to avoid leaking threads, network clients, etc. across
broker
+ * restart cycles.
+ */
+ @Override
+ public void close() throws Exception {
+ if (plugin.isPresent()) {
+ plugin.get().close();
+ }
+ }
+
+ /**
+ * @return true if a topology description plugin is configured on this
broker.
+ */
+ public boolean isPluginConfigured() {
+ return plugin.isPresent();
+ }
+
+ /**
+ * Post-processes a successful streams group heartbeat result by deciding
whether the
+ * broker should set {@code TopologyDescriptionRequired=true} on the
response, and
+ * arming the per-group back-off when it does.
+ *
+ * <p>The flag is set when the request is at a version that carries the
field
+ * ({@code TopologyDescriptionRequired} arrives at v1), the topology
description plugin
+ * is configured, the group has resolved to a topology epoch, that epoch
is neither
+ * stored nor permanently failed at the plugin, no back-off is in effect
for this
+ * epoch, and the response does not carry a {@code STALE_TOPOLOGY} status
(the member
+ * would just be told to catch up first). When the response already
carries an error
+ * code we leave it alone.
+ *
+ * <p>The version gate is intentional: a v0 client cannot deserialize the
flag, so
+ * arming the back-off for it would accumulate entries that grow
exponentially while
+ * the flag itself gets dropped at serialization — wasting heap on a
per-group basis
+ * for clients that will never push.
+ */
+ public StreamsGroupHeartbeatResult maybeSetTopologyDescriptionRequired(
+ StreamsGroupHeartbeatResult result,
+ String groupId,
+ int apiVersion
+ ) {
+ if (apiVersion < 1 || plugin.isEmpty()) {
+ return result;
+ }
+ StreamsGroupHeartbeatResponseData response = result.data();
+ if (response.errorCode() != Errors.NONE.code()) {
+ return result;
+ }
+ int currentEpoch = result.currentTopologyEpoch();
+ if (currentEpoch < 0
+ || result.storedDescriptionTopologyEpoch() == currentEpoch
+ || result.failedDescriptionTopologyEpoch() == currentEpoch
+ || responseHasStaleTopology(response)) {
+ return result;
+ }
+ // Atomic check-and-arm: only set the flag if the back-off window is
not already
+ // in effect for this epoch, so two concurrent heartbeats for the same
group cannot
+ // both arm the back-off and double the window beyond its intended
length.
+ if (backoff.armIfNotActive(groupId, currentEpoch)) {
+ response.setTopologyDescriptionRequired(true);
+ }
+ return result;
+ }
+
+ // Visible for testing.
+ StreamsGroupTopologyDescriptionBackoff backoff() {
+ return backoff;
+ }
+
+ private static boolean
responseHasStaleTopology(StreamsGroupHeartbeatResponseData response) {
+ if (response.status() == null) {
+ return false;
+ }
+ byte staleCode = Status.STALE_TOPOLOGY.code();
+ return response.status().stream().anyMatch(s -> s.statusCode() ==
staleCode);
+ }
+}
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
index 1ac1baf0553..33a71646068 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
@@ -92,6 +92,7 @@ import org.apache.kafka.common.utils.internals.LogContext;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime;
import org.apache.kafka.coordinator.common.runtime.MetadataImageBuilder;
+import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
import org.apache.kafka.coordinator.group.streams.StreamsGroupDescribeResult;
import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
@@ -473,6 +474,8 @@ public class GroupCoordinatorServiceTest {
new StreamsGroupHeartbeatResult(
new
StreamsGroupHeartbeatResponseData().setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code()),
Map.of(),
+ -1,
+ -1,
-1
),
future.get()
@@ -505,6 +508,8 @@ public class GroupCoordinatorServiceTest {
new StreamsGroupHeartbeatResult(
new StreamsGroupHeartbeatResponseData(),
Map.of(),
+ -1,
+ -1,
-1
)
));
@@ -514,7 +519,7 @@ public class GroupCoordinatorServiceTest {
request
);
- assertEquals(new StreamsGroupHeartbeatResult(new
StreamsGroupHeartbeatResponseData(), Map.of(), -1), future.get(5,
TimeUnit.SECONDS));
+ assertEquals(new StreamsGroupHeartbeatResult(new
StreamsGroupHeartbeatResponseData(), Map.of(), -1, -1, -1), future.get(5,
TimeUnit.SECONDS));
}
private static Stream<Arguments>
testStreamsGroupHeartbeatWithExceptionSource() {
@@ -574,6 +579,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(expectedErrorCode)
.setErrorMessage(expectedErrorMessage),
Map.of(),
+ -1,
+ -1,
-1
),
future.get(5, TimeUnit.SECONDS)
@@ -596,6 +603,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("TaskOffsets are not supported yet."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -611,6 +620,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("TaskEndOffsets are not supported yet."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -626,6 +637,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("WarmupTasks are not supported yet."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -641,6 +654,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("Regular expressions for source topics
are not supported yet."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -675,6 +690,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("MemberId can't be empty."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -690,6 +707,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("MemberId can't be empty."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -706,6 +725,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("GroupId can't be empty."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -722,6 +743,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("GroupId can't be empty."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -739,6 +762,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("RebalanceTimeoutMs must be provided in
first request."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -757,6 +782,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("ActiveTasks must be empty when
(re-)joining."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -776,6 +803,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("StandbyTasks must be empty when
(re-)joining."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -796,6 +825,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("WarmupTasks must be empty when
(re-)joining."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -817,6 +848,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("Topology must be non-null when
(re-)joining."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -839,6 +872,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("RackId can't be empty."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -858,6 +893,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("InstanceId can't be null."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -881,6 +918,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("MemberEpoch is -3, but must be greater
than or equal to -2."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -900,6 +939,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.INVALID_REQUEST.code())
.setErrorMessage("Topology can only be provided when
(re-)joining."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -923,6 +964,8 @@ public class GroupCoordinatorServiceTest {
.setErrorCode(Errors.STREAMS_INVALID_TOPOLOGY.code())
.setErrorMessage("Changelog topic
changelog_topic_with_fixed_partition must have an undefined partition count,
but it is set to 3."),
Map.of(),
+ -1,
+ -1,
-1
),
service.streamsGroupHeartbeat(
@@ -5922,6 +5965,7 @@ public class GroupCoordinatorServiceTest {
private Persister persister = new NoOpStatePersister();
private MetadataImage metadataImage = null;
private PartitionMetadataClient partitionMetadataClient = null;
+ private Optional<StreamsGroupTopologyDescriptionPlugin>
streamsGroupTopologyDescriptionPlugin = Optional.empty();
GroupCoordinatorService build() {
return build(false);
@@ -5934,6 +5978,7 @@ public class GroupCoordinatorServiceTest {
.build();
}
+ MockTimer mockTimer = new MockTimer();
var service = new GroupCoordinatorService(
logContext,
config,
@@ -5941,8 +5986,10 @@ public class GroupCoordinatorServiceTest {
metrics,
configManager,
persister,
- new MockTimer(),
- partitionMetadataClient
+ mockTimer,
+ partitionMetadataClient,
+ streamsGroupTopologyDescriptionPlugin,
+ mockTimer.time()
);
if (serviceStartup) {
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
new file mode 100644
index 00000000000..7ed02b37d92
--- /dev/null
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
@@ -0,0 +1,285 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.internals.Topic;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
+import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.internals.LogContext;
+import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
+import org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime;
+import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin;
+import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
+import org.apache.kafka.server.share.persister.NoOpStatePersister;
+import org.apache.kafka.server.util.timer.MockTimer;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import static
org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse.Status;
+import static
org.apache.kafka.coordinator.common.runtime.TestUtil.requestContext;
+import static
org.apache.kafka.coordinator.group.GroupConfigManagerTest.createConfigManager;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for the heartbeat post-processing on {@link GroupCoordinatorService}
that sets
+ * {@code TopologyDescriptionRequired} on the response when a topology
description plugin
+ * is configured and the stored/failed epochs lag the current topology epoch.
+ */
+public class GroupCoordinatorServiceTopologyDescriptionTest {
+
+ private static final TopicPartition GROUP_TP = new
TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 0);
+
+ @SuppressWarnings("unchecked")
+ private static CoordinatorRuntime<GroupCoordinatorShard,
CoordinatorRecord> mockRuntime() {
+ return (CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord>)
mock(CoordinatorRuntime.class);
+ }
+
+ private static GroupCoordinatorService buildService(
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime,
+ Optional<StreamsGroupTopologyDescriptionPlugin> plugin,
+ boolean startup
+ ) {
+ return buildService(runtime, plugin, startup, new MockTimer());
+ }
+
+ private static GroupCoordinatorService buildService(
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime,
+ Optional<StreamsGroupTopologyDescriptionPlugin> plugin,
+ boolean startup,
+ MockTimer timer
+ ) {
+ MockTime time = timer.time();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ GroupCoordinatorConfigTest.createGroupCoordinatorConfig(4096,
600000L, 24),
+ runtime,
+ new GroupCoordinatorMetrics(),
+ createConfigManager(),
+ new NoOpStatePersister(),
+ timer,
+ null,
+ plugin,
+ time
+ );
+ if (startup) {
+ service.startup(() -> 1);
+ }
+ return service;
+ }
+
+ @Test
+ public void testHeartbeatSetsTopologyDescriptionRequiredWhenStoredLags()
throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(runtime.scheduleWriteOperation(
+ eq("streams-group-heartbeat"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupHeartbeatResult(new
StreamsGroupHeartbeatResponseData(), Map.of(), 5, -1, -1)));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ StreamsGroupHeartbeatResult result = service.streamsGroupHeartbeat(
+ requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT),
validHeartbeatRequest()
+ ).get(5, TimeUnit.SECONDS);
+
+ assertTrue(result.data().topologyDescriptionRequired());
+ }
+
+ @Test
+ public void testHeartbeatArmSuppressReSolicitCycle() throws Exception {
+ // End-to-end exercise of the arm → suppress → re-solicit cycle
through the
+ // service + TopologyDescriptionManager. Backoff primitive tests cover
this in
+ // isolation; this asserts the contract holds when the heartbeat write
+ // result is fed into maybeSetTopologyDescriptionRequired.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ // Each call must yield a fresh response so mutations from the
previous call do not
+ // bleed through. thenReturn would hand back the same instance to
every invocation
+ // and the second heartbeat would observe topologyDescriptionRequired
carried over
+ // from the first, masking the suppression we want to assert.
+ when(runtime.scheduleWriteOperation(
+ eq("streams-group-heartbeat"),
+ eq(GROUP_TP),
+ any()
+ )).thenAnswer(invocation -> CompletableFuture.completedFuture(
+ new StreamsGroupHeartbeatResult(new
StreamsGroupHeartbeatResponseData(), Map.of(), 5, -1, -1)));
+
+ MockTimer timer = new MockTimer();
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true, timer);
+
+ // 1. First heartbeat — back-off idle. Manager arms it and sets the
flag.
+ StreamsGroupHeartbeatResult firstResult =
service.streamsGroupHeartbeat(
+ requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT),
validHeartbeatRequest()
+ ).get(5, TimeUnit.SECONDS);
+ assertTrue(firstResult.data().topologyDescriptionRequired());
+
+ // 2. Second heartbeat — back-off window still active. Manager
suppresses the flag.
+ StreamsGroupHeartbeatResult secondResult =
service.streamsGroupHeartbeat(
+ requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT),
validHeartbeatRequest()
+ ).get(5, TimeUnit.SECONDS);
+ assertFalse(secondResult.data().topologyDescriptionRequired());
+
+ // 3. Advance MockTime past the back-off window. INITIAL_DELAY_MS is
30s — the
+ // value lives in StreamsGroupTopologyDescriptionBackoff as a
package-private
+ // constant; sleeping a comfortable margin past it keeps this test
independent
+ // of the exact delay while still asserting "past the initial window".
+ timer.time().sleep(60_000L);
+
+ // 4. Third heartbeat — window expired. Manager re-arms and sets the
flag again.
+ StreamsGroupHeartbeatResult thirdResult =
service.streamsGroupHeartbeat(
+ requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT),
validHeartbeatRequest()
+ ).get(5, TimeUnit.SECONDS);
+ assertTrue(thirdResult.data().topologyDescriptionRequired());
+ }
+
+ @Test
+ public void testHeartbeatSkipsFlagWhenStoredMatchesCurrent() throws
Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(runtime.scheduleWriteOperation(
+ eq("streams-group-heartbeat"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupHeartbeatResult(new
StreamsGroupHeartbeatResponseData(), Map.of(), 5, 5, -1)));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ StreamsGroupHeartbeatResult result = service.streamsGroupHeartbeat(
+ requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT),
validHeartbeatRequest()
+ ).get(5, TimeUnit.SECONDS);
+
+ assertFalse(result.data().topologyDescriptionRequired());
+ }
+
+ @Test
+ public void testHeartbeatSkipsFlagWhenFailedAtCurrentEpoch() throws
Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(runtime.scheduleWriteOperation(
+ eq("streams-group-heartbeat"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupHeartbeatResult(new
StreamsGroupHeartbeatResponseData(), Map.of(), 5, -1, 5)));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ StreamsGroupHeartbeatResult result = service.streamsGroupHeartbeat(
+ requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT),
validHeartbeatRequest()
+ ).get(5, TimeUnit.SECONDS);
+
+ assertFalse(result.data().topologyDescriptionRequired());
+ }
+
+ @Test
+ public void testHeartbeatSkipsFlagWhenStaleTopologyStatusPresent() throws
Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ StreamsGroupHeartbeatResponseData responseData = new
StreamsGroupHeartbeatResponseData()
+ .setStatus(List.of(new StreamsGroupHeartbeatResponseData.Status()
+ .setStatusCode(Status.STALE_TOPOLOGY.code())
+ .setStatusDetail("behind")));
+ when(runtime.scheduleWriteOperation(
+ eq("streams-group-heartbeat"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupHeartbeatResult(responseData, Map.of(), 5, -1,
-1)));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ StreamsGroupHeartbeatResult result = service.streamsGroupHeartbeat(
+ requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT),
validHeartbeatRequest()
+ ).get(5, TimeUnit.SECONDS);
+
+ assertFalse(result.data().topologyDescriptionRequired());
+ }
+
+ @Test
+ public void testHeartbeatNeverSetsFlagWithoutPlugin() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ when(runtime.scheduleWriteOperation(
+ eq("streams-group-heartbeat"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupHeartbeatResult(new
StreamsGroupHeartbeatResponseData(), Map.of(), 5, -1, -1)));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.empty(), true);
+ StreamsGroupHeartbeatResult result = service.streamsGroupHeartbeat(
+ requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT),
validHeartbeatRequest()
+ ).get(5, TimeUnit.SECONDS);
+
+ assertFalse(result.data().topologyDescriptionRequired());
+ }
+
+ @Test
+ public void testHeartbeatSkipsFlagOnV0Request() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(runtime.scheduleWriteOperation(
+ eq("streams-group-heartbeat"),
+ eq(GROUP_TP),
+ any()
+ )).thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupHeartbeatResult(new
StreamsGroupHeartbeatResponseData(), Map.of(), 5, -1, -1)));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ StreamsGroupHeartbeatResult result = service.streamsGroupHeartbeat(
+ requestContext(ApiKeys.STREAMS_GROUP_HEARTBEAT, (short) 0),
validHeartbeatRequest()
+ ).get(5, TimeUnit.SECONDS);
+
+ assertFalse(result.data().topologyDescriptionRequired());
+ }
+
+ @Test
+ public void testShutdownClosesPlugin() throws Exception {
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ GroupCoordinatorService service = buildService(mockRuntime(),
Optional.of(plugin), true);
+ service.shutdown();
+ verify(plugin, times(1)).close();
+ }
+
+ private static StreamsGroupHeartbeatRequestData validHeartbeatRequest() {
+ return new StreamsGroupHeartbeatRequestData()
+ .setGroupId("foo")
+ .setMemberId(Uuid.randomUuid().toString())
+ .setMemberEpoch(0)
+ .setRebalanceTimeoutMs(1500)
+ .setActiveTasks(List.of())
+ .setStandbyTasks(List.of())
+ .setWarmupTasks(List.of())
+ .setTopology(new StreamsGroupHeartbeatRequestData.Topology());
+ }
+}
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
index 72e2d9d11f7..448145c54fa 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorShardTest.java
@@ -195,7 +195,7 @@ public class GroupCoordinatorShardTest {
StreamsGroupHeartbeatRequestData request = new
StreamsGroupHeartbeatRequestData();
CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord>
result = new CoordinatorResult<>(
List.of(),
- new StreamsGroupHeartbeatResult(new
StreamsGroupHeartbeatResponseData(), Map.of(), -1)
+ new StreamsGroupHeartbeatResult(new
StreamsGroupHeartbeatResponseData(), Map.of(), -1, -1, -1)
);
when(groupMetadataManager.streamsGroupHeartbeat(
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResultTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResultTest.java
index e08fcd36d2e..2fc3f23e007 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResultTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupHeartbeatResultTest.java
@@ -25,35 +25,47 @@ import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
public class StreamsGroupHeartbeatResultTest {
@Test
- public void testThreeArgConstructorPreservesTopologyEpoch() {
+ public void testConstructorPreservesEpochs() {
StreamsGroupHeartbeatResult result = new StreamsGroupHeartbeatResult(
- new StreamsGroupHeartbeatResponseData(), Map.of(), 7);
+ new StreamsGroupHeartbeatResponseData(), Map.of(), 7, 5, 3);
assertEquals(7, result.currentTopologyEpoch());
+ assertEquals(5, result.storedDescriptionTopologyEpoch());
+ assertEquals(3, result.failedDescriptionTopologyEpoch());
}
@Test
public void testCurrentTopologyEpochIsPartOfEquality() {
- // Records derive equals from all components; two results with
different topology epochs are unequal.
StreamsGroupHeartbeatResult a = new StreamsGroupHeartbeatResult(
- new StreamsGroupHeartbeatResponseData(), Map.of(), 1);
+ new StreamsGroupHeartbeatResponseData(), Map.of(), 1, -1, -1);
StreamsGroupHeartbeatResult b = new StreamsGroupHeartbeatResult(
- new StreamsGroupHeartbeatResponseData(), Map.of(), 2);
+ new StreamsGroupHeartbeatResponseData(), Map.of(), 2, -1, -1);
assertNotEquals(a, b);
StreamsGroupHeartbeatResult c = new StreamsGroupHeartbeatResult(
- new StreamsGroupHeartbeatResponseData(), Map.of(), 1);
+ new StreamsGroupHeartbeatResponseData(), Map.of(), 1, -1, -1);
assertEquals(a, c);
}
+ @Test
+ public void testStoredAndFailedEpochsArePartOfEquality() {
+ StreamsGroupHeartbeatResult a = new StreamsGroupHeartbeatResult(
+ new StreamsGroupHeartbeatResponseData(), Map.of(), 1, 1, -1);
+ StreamsGroupHeartbeatResult differentStored = new
StreamsGroupHeartbeatResult(
+ new StreamsGroupHeartbeatResponseData(), Map.of(), 1, 0, -1);
+ StreamsGroupHeartbeatResult differentFailed = new
StreamsGroupHeartbeatResult(
+ new StreamsGroupHeartbeatResponseData(), Map.of(), 1, 1, 0);
+ assertNotEquals(a, differentStored);
+ assertNotEquals(a, differentFailed);
+ }
+
@Test
public void testCreatableTopicsMapIsImmutable() {
StreamsGroupHeartbeatResult result = new StreamsGroupHeartbeatResult(
- new StreamsGroupHeartbeatResponseData(), Map.of(), -1);
+ new StreamsGroupHeartbeatResponseData(), Map.of(), -1, -1, -1);
assertThrows(UnsupportedOperationException.class,
() -> result.creatableTopics().put("t", null));
}
@@ -61,7 +73,6 @@ public class StreamsGroupHeartbeatResultTest {
@Test
public void testNullDataIsRejected() {
assertThrows(NullPointerException.class,
- () -> new StreamsGroupHeartbeatResult(null, Map.of(), -1));
- assertTrue(true);
+ () -> new StreamsGroupHeartbeatResult(null, Map.of(), -1, -1, -1));
}
}
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionBackoffTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionBackoffTest.java
new file mode 100644
index 00000000000..d6eae64a3e8
--- /dev/null
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionBackoffTest.java
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams;
+
+import org.apache.kafka.common.utils.MockTime;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class StreamsGroupTopologyDescriptionBackoffTest {
+
+ @Test
+ public void testFirstArmUsesInitialDelay() {
+ MockTime time = new MockTime();
+ StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ backoff.armOrExtend("g", 1);
+ assertTrue(backoff.isActive("g", 1));
+ time.sleep(StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS -
1);
+ assertTrue(backoff.isActive("g", 1));
+ time.sleep(1);
+ assertFalse(backoff.isActive("g", 1));
+ }
+
+ @Test
+ public void testConsecutiveArmsDoubleTheWindowUpToMax() {
+ MockTime time = new MockTime();
+ StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+
+ long expected =
StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS;
+ backoff.armOrExtend("g", 1);
+ assertEquals(expected, backoff.entry("g").currentDelayMs());
+
+ // Each consecutive arm at the same epoch doubles, until we hit the
cap.
+ for (int i = 0; i < 20; i++) {
+ backoff.armOrExtend("g", 1);
+ expected = Math.min(expected * 2,
StreamsGroupTopologyDescriptionBackoff.MAX_DELAY_MS);
+ assertEquals(expected, backoff.entry("g").currentDelayMs(),
+ "iteration " + i);
+ }
+ assertEquals(StreamsGroupTopologyDescriptionBackoff.MAX_DELAY_MS,
+ backoff.entry("g").currentDelayMs());
+ }
+
+ @Test
+ public void testDifferentEpochResetsTheWindow() {
+ MockTime time = new MockTime();
+ StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ backoff.armOrExtend("g", 1);
+ backoff.armOrExtend("g", 1); // doubled
+ long doubled = backoff.entry("g").currentDelayMs();
+ assertTrue(doubled >
StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS);
+ backoff.armOrExtend("g", 2);
+ assertEquals(StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS,
+ backoff.entry("g").currentDelayMs());
+ assertEquals(2, backoff.entry("g").topologyEpoch());
+ }
+
+ @Test
+ public void testIsActiveIsScopedToEpoch() {
+ MockTime time = new MockTime();
+ StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ backoff.armOrExtend("g", 1);
+ assertTrue(backoff.isActive("g", 1));
+ // A query at a different epoch never matches — the broker should
re-solicit.
+ assertFalse(backoff.isActive("g", 2));
+ }
+
+ @Test
+ public void testClearRemovesEntry() {
+ MockTime time = new MockTime();
+ StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ backoff.armOrExtend("g", 1);
+ assertNotNull(backoff.entry("g"));
+ backoff.clear("g");
+ assertNull(backoff.entry("g"));
+ assertFalse(backoff.isActive("g", 1));
+ }
+
+ @Test
+ public void testArmIfNotActiveArmsWhenIdle() {
+ MockTime time = new MockTime();
+ StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ assertTrue(backoff.armIfNotActive("g", 1));
+ assertTrue(backoff.isActive("g", 1));
+ assertEquals(StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS,
+ backoff.entry("g").currentDelayMs());
+ }
+
+ @Test
+ public void testArmIfNotActiveReturnsFalseWhenAlreadyActive() {
+ MockTime time = new MockTime();
+ StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ assertTrue(backoff.armIfNotActive("g", 1));
+ long armedAt = backoff.entry("g").nextAttemptMs();
+ // A second call inside the window does not arm and does not extend
the window.
+ assertFalse(backoff.armIfNotActive("g", 1));
+ assertEquals(armedAt, backoff.entry("g").nextAttemptMs());
+ }
+
+ @Test
+ public void testArmIfNotActiveReArmsAfterEpochAdvance() {
+ MockTime time = new MockTime();
+ StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+ assertTrue(backoff.armIfNotActive("g", 1));
+ // A query at the new epoch sees no active window and arms a fresh one.
+ assertTrue(backoff.armIfNotActive("g", 2));
+ assertEquals(2, backoff.entry("g").topologyEpoch());
+ assertEquals(StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS,
+ backoff.entry("g").currentDelayMs());
+ }
+
+ @Test
+ public void testArmIfNotActiveDoublesAfterExpiredWindowAtSameEpoch() {
+ // Heartbeats keep re-soliciting the same epoch because the client
never
+ // pushed (or the push was lost). The exponential chain must continue
across
+ // those re-arms instead of resetting to INITIAL_DELAY_MS every cycle.
+ MockTime time = new MockTime();
+ StreamsGroupTopologyDescriptionBackoff backoff = new
StreamsGroupTopologyDescriptionBackoff(time);
+
+ long expected =
StreamsGroupTopologyDescriptionBackoff.INITIAL_DELAY_MS;
+ assertTrue(backoff.armIfNotActive("g", 1));
+ assertEquals(expected, backoff.entry("g").currentDelayMs());
+
+ for (int i = 0; i < 20; i++) {
+ time.sleep(backoff.entry("g").currentDelayMs());
+ assertTrue(backoff.armIfNotActive("g", 1));
+ expected = Math.min(expected * 2,
StreamsGroupTopologyDescriptionBackoff.MAX_DELAY_MS);
+ assertEquals(expected, backoff.entry("g").currentDelayMs(),
"iteration " + i);
+ }
+ assertEquals(StreamsGroupTopologyDescriptionBackoff.MAX_DELAY_MS,
+ backoff.entry("g").currentDelayMs());
+ }
+}
