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 a3362c491e9 KAFKA-20625: Add
StreamsGroupTopologyDescriptionRequestManager and Streams client topology push
[2/N] (#22640)
a3362c491e9 is described below
commit a3362c491e9d36b4dcc4827232d3e11af50a20c4
Author: Lucy Liu <[email protected]>
AuthorDate: Wed Jun 24 14:16:34 2026 -0500
KAFKA-20625: Add StreamsGroupTopologyDescriptionRequestManager and Streams
client topology push [2/N] (#22640)
## Summary
Adds the client-side request manager that pushes topology descriptions
to the broker, wires the heartbeat path to flip the push-required flag
on responses, and registers the new manager in the consumer network
thread. Should be based on https://github.com/apache/kafka/pull/22639
## File changes
- **`StreamsGroupTopologyDescriptionRequestManager`**: new
RequestManager that implements the KIP-1331 response handling with unit
tests.
- **`StreamsGroupHeartbeatRequestManager`**: `onSuccessResponse`
propagates `memberId` and sets `topologyPushRequired` when the broker
requests a push, with unit tests.
- **`RequestManagers`**: registers the new manager in the Streams branch
and the entries list. Current tests updated.
- **`StreamThread.initStreamsRebalanceData`**: bumped to package-private
for testing. 2 new tests verify the wire description is
populated/skipped based on the config.
Reviewers: Lucas Brutschy <[email protected]>
---------
Co-authored-by: Lucas Brutschy <[email protected]>
---
.../consumer/internals/RequestManagers.java | 17 +
.../StreamsGroupHeartbeatRequestManager.java | 6 +
...eamsGroupTopologyDescriptionRequestManager.java | 181 +++++++++++
.../consumer/internals/RequestManagersTest.java | 8 +-
.../StreamsGroupHeartbeatRequestManagerTest.java | 76 +++++
...GroupTopologyDescriptionRequestManagerTest.java | 342 +++++++++++++++++++++
.../events/ApplicationEventProcessorTest.java | 4 +
.../streams/processor/internals/StreamThread.java | 5 +-
.../processor/internals/StreamThreadTest.java | 46 +++
9 files changed, 681 insertions(+), 4 deletions(-)
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
index 36ebbb3bbf3..e93da12cda6 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java
@@ -61,6 +61,7 @@ public class RequestManagers implements Closeable {
public final FetchRequestManager fetchRequestManager;
public final Optional<ShareConsumeRequestManager>
shareConsumeRequestManager;
public final Optional<StreamsGroupHeartbeatRequestManager>
streamsGroupHeartbeatRequestManager;
+ public final Optional<StreamsGroupTopologyDescriptionRequestManager>
streamsGroupTopologyDescriptionRequestManager;
private final List<RequestManager> entries;
private final IdempotentCloser closer = new IdempotentCloser();
@@ -73,6 +74,7 @@ public class RequestManagers implements Closeable {
Optional<ConsumerHeartbeatRequestManager>
heartbeatRequestManager,
Optional<ConsumerMembershipManager>
membershipManager,
Optional<StreamsGroupHeartbeatRequestManager>
streamsGroupHeartbeatRequestManager,
+
Optional<StreamsGroupTopologyDescriptionRequestManager>
streamsGroupTopologyDescriptionRequestManager,
Optional<StreamsMembershipManager>
streamsMembershipManager) {
this.log = logContext.logger(RequestManagers.class);
this.offsetsRequestManager = requireNonNull(offsetsRequestManager,
"OffsetsRequestManager cannot be null");
@@ -84,6 +86,7 @@ public class RequestManagers implements Closeable {
this.consumerHeartbeatRequestManager = heartbeatRequestManager;
this.shareHeartbeatRequestManager = Optional.empty();
this.streamsGroupHeartbeatRequestManager =
streamsGroupHeartbeatRequestManager;
+ this.streamsGroupTopologyDescriptionRequestManager =
streamsGroupTopologyDescriptionRequestManager;
this.consumerMembershipManager = membershipManager;
this.streamsMembershipManager = streamsMembershipManager;
this.shareMembershipManager = Optional.empty();
@@ -94,6 +97,7 @@ public class RequestManagers implements Closeable {
heartbeatRequestManager.ifPresent(list::add);
membershipManager.ifPresent(list::add);
streamsGroupHeartbeatRequestManager.ifPresent(list::add);
+ streamsGroupTopologyDescriptionRequestManager.ifPresent(list::add);
streamsMembershipManager.ifPresent(list::add);
list.add(offsetsRequestManager);
list.add(topicMetadataRequestManager);
@@ -112,6 +116,7 @@ public class RequestManagers implements Closeable {
this.commitRequestManager = Optional.empty();
this.consumerHeartbeatRequestManager = Optional.empty();
this.streamsGroupHeartbeatRequestManager = Optional.empty();
+ this.streamsGroupTopologyDescriptionRequestManager = Optional.empty();
this.shareHeartbeatRequestManager = shareHeartbeatRequestManager;
this.consumerMembershipManager = Optional.empty();
this.streamsMembershipManager = Optional.empty();
@@ -199,6 +204,7 @@ public class RequestManagers implements Closeable {
CoordinatorRequestManager coordinator = null;
CommitRequestManager commitRequestManager = null;
StreamsGroupHeartbeatRequestManager
streamsGroupHeartbeatRequestManager = null;
+ StreamsGroupTopologyDescriptionRequestManager
streamsGroupTopologyDescriptionRequestManager = null;
StreamsMembershipManager streamsMembershipManager = null;
if (groupRebalanceConfig != null &&
groupRebalanceConfig.groupId != null) {
@@ -247,6 +253,16 @@ public class RequestManagers implements Closeable {
metrics,
streamsRebalanceData.get()
);
+
+ streamsGroupTopologyDescriptionRequestManager = new
StreamsGroupTopologyDescriptionRequestManager(
+ logContext,
+ time,
+ retryBackoffMs,
+ retryBackoffMaxMs,
+ streamsMembershipManager,
+ streamsRebalanceData.get(),
+ coordinator
+ );
} else {
membershipManager = new ConsumerMembershipManager(
groupRebalanceConfig.groupId,
@@ -308,6 +324,7 @@ public class RequestManagers implements Closeable {
Optional.ofNullable(heartbeatRequestManager),
Optional.ofNullable(membershipManager),
Optional.ofNullable(streamsGroupHeartbeatRequestManager),
+
Optional.ofNullable(streamsGroupTopologyDescriptionRequestManager),
Optional.ofNullable(streamsMembershipManager)
);
}
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java
index 4793025f88c..fce71fe95d5 100644
---
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java
@@ -571,6 +571,10 @@ public class StreamsGroupHeartbeatRequestManager
implements RequestManager {
streamsRebalanceData.setTaskOffsetIntervalMs(data.taskOffsetIntervalMs());
streamsRebalanceData.setAcceptableRecoveryLag(data.acceptableRecoveryLag());
+ if (data.topologyDescriptionRequired() &&
streamsRebalanceData.wireTopologyDescription() != null) {
+ streamsRebalanceData.setTopologyPushRequired(true);
+ }
+
if (data.partitionsByUserEndpoint() != null) {
streamsRebalanceData.setPartitionsByHost(convertHostInfoMap(data));
}
@@ -673,6 +677,7 @@ public class StreamsGroupHeartbeatRequestManager implements
RequestManager {
membershipManager.onFenced();
// Skip backoff so that a next HB to rejoin is sent as soon as
the fenced member releases its assignment
heartbeatRequestState.reset();
+ streamsRebalanceData.setTopologyPushRequired(false);
break;
case UNKNOWN_MEMBER_ID:
@@ -685,6 +690,7 @@ public class StreamsGroupHeartbeatRequestManager implements
RequestManager {
membershipManager.onFenced();
// Skip backoff so that a next HB to rejoin is sent as soon as
the fenced member releases its assignment
heartbeatRequestState.reset();
+ streamsRebalanceData.setTopologyPushRequired(false);
break;
case UNSUPPORTED_VERSION:
diff --git
a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupTopologyDescriptionRequestManager.java
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupTopologyDescriptionRequestManager.java
new file mode 100644
index 00000000000..30979905693
--- /dev/null
+++
b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupTopologyDescriptionRequestManager.java
@@ -0,0 +1,181 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.errors.RetriableException;
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
+import org.apache.kafka.common.protocol.Errors;
+import
org.apache.kafka.common.requests.StreamsGroupTopologyDescriptionUpdateRequest;
+import
org.apache.kafka.common.requests.StreamsGroupTopologyDescriptionUpdateResponse;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.internals.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.Collections;
+import java.util.Objects;
+
+public class StreamsGroupTopologyDescriptionRequestManager implements
RequestManager {
+
+ private final Logger logger;
+ private final Time time;
+ private final StreamsMembershipManager membershipManager;
+ private final StreamsRebalanceData streamsRebalanceData;
+ private final CoordinatorRequestManager coordinatorRequestManager;
+ private final RequestState pushRequestState;
+
+ private long nextPushTimeMs = 0L;
+
+ public StreamsGroupTopologyDescriptionRequestManager(final LogContext
logContext,
+ final Time time,
+ final long
retryBackoffMs,
+ final long
retryBackoffMaxMs,
+ final
StreamsMembershipManager membershipManager,
+ final
StreamsRebalanceData streamsRebalanceData,
+ final
CoordinatorRequestManager coordinatorRequestManager) {
+ this.logger = logContext.logger(getClass());
+ this.time = Objects.requireNonNull(time);
+ this.membershipManager = Objects.requireNonNull(membershipManager);
+ this.streamsRebalanceData =
Objects.requireNonNull(streamsRebalanceData);
+ this.coordinatorRequestManager =
Objects.requireNonNull(coordinatorRequestManager);
+ this.pushRequestState = new RequestState(
+ logContext,
+
StreamsGroupTopologyDescriptionRequestManager.class.getSimpleName(),
+ retryBackoffMs,
+ retryBackoffMaxMs);
+ }
+
+ @Override
+ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
+ if (!shouldSendTopologyDescriptionUpdate(currentTimeMs)) {
+ return NetworkClientDelegate.PollResult.EMPTY;
+ }
+
+ final StreamsGroupTopologyDescriptionUpdateRequestData data = new
StreamsGroupTopologyDescriptionUpdateRequestData()
+ .setGroupId(membershipManager.groupId())
+ .setMemberId(membershipManager.memberId())
+ .setTopologyEpoch(streamsRebalanceData.topologyEpoch())
+
.setTopologyDescription(streamsRebalanceData.wireTopologyDescription());
+
+ final NetworkClientDelegate.UnsentRequest unsent = new
NetworkClientDelegate.UnsentRequest(
+ new StreamsGroupTopologyDescriptionUpdateRequest.Builder(data),
+ coordinatorRequestManager.coordinator()
+ );
+ unsent.whenComplete((response, exception) -> onResponse(response,
exception));
+
+ pushRequestState.onSendAttempt(currentTimeMs);
+ return new
NetworkClientDelegate.PollResult(Collections.singletonList(unsent));
+ }
+
+ @Override
+ public long maximumTimeToWait(final long currentTimeMs) {
+ if (!streamsRebalanceData.topologyPushRequired()) {
+ return Long.MAX_VALUE;
+ }
+ final long backoffRemainingMs =
pushRequestState.remainingBackoffMs(currentTimeMs);
+ final long throttleRemainingMs = Math.max(0L, nextPushTimeMs -
currentTimeMs);
+ final long waitMs = Math.max(backoffRemainingMs, throttleRemainingMs);
+ if (waitMs > 0L) {
+ return waitMs;
+ }
+ return shouldSendTopologyDescriptionUpdate(currentTimeMs) ? 0L :
Long.MAX_VALUE;
+ }
+
+ private boolean shouldSendTopologyDescriptionUpdate(final long
currentTimeMs) {
+ if (!pushRequestState.canSendRequest(currentTimeMs) || currentTimeMs <
nextPushTimeMs) {
+ return false;
+ }
+ if (!streamsRebalanceData.topologyPushRequired() ||
streamsRebalanceData.wireTopologyDescription() == null) {
+ return false;
+ }
+ final String memberId = membershipManager.memberId();
+ if (memberId == null || memberId.isEmpty()) {
+ return false;
+ }
+ return coordinatorRequestManager.coordinator().isPresent();
+ }
+
+ private void onResponse(final ClientResponse response, final Throwable
exception) {
+ final long responseTimeMs = time.milliseconds();
+
+ if (exception != null) {
+ if (exception instanceof RetriableException) {
+ pushRequestState.onFailedAttempt(responseTimeMs);
+
coordinatorRequestManager.handleCoordinatorDisconnect(exception,
responseTimeMs);
+ logger.warn("Topology description push failed with retriable
exception; will retry on next poll", exception);
+ } else {
+ // Non-retriable exceptions should clear the flag and give up.
+ pushRequestState.onSuccessfulAttempt(responseTimeMs);
+ streamsRebalanceData.setTopologyPushRequired(false);
+ logger.warn("Topology description push failed with
non-retriable exception.", exception);
+ }
+ return;
+ }
+
+ final StreamsGroupTopologyDescriptionUpdateResponse body =
+ (StreamsGroupTopologyDescriptionUpdateResponse)
response.responseBody();
+ final Errors error = Errors.forCode(body.data().errorCode());
+ final String errorMessage = body.data().errorMessage();
+
+ if (body.data().throttleTimeMs() > 0) {
+ nextPushTimeMs = responseTimeMs + body.data().throttleTimeMs();
+ }
+
+ switch (error) {
+ case NONE:
+ pushRequestState.onSuccessfulAttempt(responseTimeMs);
+ streamsRebalanceData.setTopologyPushRequired(false);
+ break;
+
+ case NOT_COORDINATOR:
+ case COORDINATOR_NOT_AVAILABLE:
+ pushRequestState.onFailedAttempt(responseTimeMs);
+ logger.info("Coordinator error {} pushing topology
description. Will rediscover and retry: {}", error, errorMessage);
+ coordinatorRequestManager.markCoordinatorUnknown(errorMessage,
responseTimeMs);
+ break;
+
+ case COORDINATOR_LOAD_IN_PROGRESS:
+ pushRequestState.onFailedAttempt(responseTimeMs);
+ logger.info("Coordinator is loading; will retry on next poll:
{}", errorMessage);
+ break;
+
+ case UNKNOWN_MEMBER_ID:
+ // Member was dropped — clear the flag and let the heartbeat
path drive the rejoin.
+ // onSuccessfulAttempt resets request state without backoff
since no retry follows.
+ pushRequestState.onSuccessfulAttempt(responseTimeMs);
+ logger.info("Topology description push rejected with
UNKNOWN_MEMBER_ID; heartbeat will trigger rejoin: {}", errorMessage);
+ streamsRebalanceData.setTopologyPushRequired(false);
+ break;
+
+ case STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED:
+ case INVALID_REQUEST:
+ case UNSUPPORTED_VERSION:
+ case GROUP_ID_NOT_FOUND:
+ case GROUP_AUTHORIZATION_FAILED:
+ default:
+ // Use onSuccessfulAttempt because no retry follows,
+ // a future push triggered by a later heartbeat should not
inherit backoff from this failure.
+ // the broker will re-signal via heartbeat if a push is needed
again.
+ pushRequestState.onSuccessfulAttempt(responseTimeMs);
+ logger.warn("Topology description push failed with {}: {}",
error, errorMessage);
+ streamsRebalanceData.setTopologyPushRequired(false);
+ break;
+ }
+ }
+
+}
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestManagersTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestManagersTest.java
index 63e24ca0228..cc74aee7760 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestManagersTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/RequestManagersTest.java
@@ -39,7 +39,7 @@ import static org.mockito.Mockito.mock;
public class RequestManagersTest {
@Test
- public void testMemberStateListenerRegistered() {
+ public void testConsumerGroupRequestManagersAndListenersWired() {
final MemberStateListener listener = (memberEpoch, memberId) -> { };
@@ -77,6 +77,7 @@ public class RequestManagersTest {
assertTrue(requestManagers.consumerMembershipManager.isPresent());
assertTrue(requestManagers.streamsMembershipManager.isEmpty());
assertTrue(requestManagers.streamsGroupHeartbeatRequestManager.isEmpty());
+
assertTrue(requestManagers.streamsGroupTopologyDescriptionRequestManager.isEmpty());
assertEquals(2,
requestManagers.consumerMembershipManager.get().stateListeners().size());
assertTrue(requestManagers.consumerMembershipManager.get().stateListeners().stream()
@@ -85,7 +86,7 @@ public class RequestManagersTest {
}
@Test
- public void testStreamMemberStateListenerRegistered() {
+ public void testStreamsGroupRequestManagersAndListenersWired() {
final MemberStateListener listener = (memberEpoch, memberId) -> { };
@@ -122,6 +123,9 @@ public class RequestManagersTest {
).get();
assertTrue(requestManagers.streamsMembershipManager.isPresent());
assertTrue(requestManagers.streamsGroupHeartbeatRequestManager.isPresent());
+
assertTrue(requestManagers.streamsGroupTopologyDescriptionRequestManager.isPresent());
+ assertTrue(requestManagers.entries().stream()
+ .anyMatch(rm -> rm instanceof
StreamsGroupTopologyDescriptionRequestManager));
assertTrue(requestManagers.consumerMembershipManager.isEmpty());
assertEquals(2,
requestManagers.streamsMembershipManager.get().stateListeners().size());
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java
index 9d92d251745..c2083fbafd0 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java
@@ -29,6 +29,7 @@ import
org.apache.kafka.common.errors.TopicAuthorizationException;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
@@ -1712,6 +1713,62 @@ class StreamsGroupHeartbeatRequestManagerTest {
}
}
+ @Test
+ public void
testStreamsRebalanceDataTopologyPushRequiredSetOnRequiredTrue() {
+ try (
+ final MockedConstruction<HeartbeatRequestState> ignored =
mockConstruction(
+ HeartbeatRequestState.class,
+ (mock, context) ->
when(mock.canSendRequest(time.milliseconds())).thenReturn(true))
+ ) {
+ final StreamsGroupHeartbeatRequestManager heartbeatRequestManager
= createStreamsGroupHeartbeatRequestManager();
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(coordinatorNode));
+ when(membershipManager.groupId()).thenReturn(GROUP_ID);
+ when(membershipManager.memberId()).thenReturn(MEMBER_ID);
+ when(membershipManager.memberEpoch()).thenReturn(MEMBER_EPOCH);
+
when(membershipManager.groupInstanceId()).thenReturn(Optional.of(INSTANCE_ID));
+
+ // Push is enabled (wire description present) so the broker's
required=true flips the flag.
+ streamsRebalanceData.setWireTopologyDescription(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription());
+ assertFalse(streamsRebalanceData.topologyPushRequired());
+
+ final NetworkClientDelegate.PollResult result =
heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(1, result.unsentRequests.size());
+
+ final NetworkClientDelegate.UnsentRequest networkRequest =
result.unsentRequests.get(0);
+
networkRequest.handler().onComplete(buildClientResponseWithTopologyRequired(true));
+
+ assertTrue(streamsRebalanceData.topologyPushRequired());
+ }
+ }
+
+ @Test
+ public void
testStreamsRebalanceDataTopologyPushRequiredNotClearedOnRequiredFalse() {
+ try (
+ final MockedConstruction<HeartbeatRequestState> ignored =
mockConstruction(
+ HeartbeatRequestState.class,
+ (mock, context) ->
when(mock.canSendRequest(time.milliseconds())).thenReturn(true))
+ ) {
+ final StreamsGroupHeartbeatRequestManager heartbeatRequestManager
= createStreamsGroupHeartbeatRequestManager();
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(coordinatorNode));
+ when(membershipManager.groupId()).thenReturn(GROUP_ID);
+ when(membershipManager.memberId()).thenReturn(MEMBER_ID);
+ when(membershipManager.memberEpoch()).thenReturn(MEMBER_EPOCH);
+
when(membershipManager.groupInstanceId()).thenReturn(Optional.of(INSTANCE_ID));
+
+ // Pre-set the flag to true to verify it survives a response that
doesn't ask for push
+ streamsRebalanceData.setTopologyPushRequired(true);
+
+ final NetworkClientDelegate.PollResult result =
heartbeatRequestManager.poll(time.milliseconds());
+ assertEquals(1, result.unsentRequests.size());
+
+ final NetworkClientDelegate.UnsentRequest networkRequest =
result.unsentRequests.get(0);
+
networkRequest.handler().onComplete(buildClientResponseWithTopologyRequired(false));
+
+ assertTrue(streamsRebalanceData.topologyPushRequired());
+ }
+ }
+
private static ConsumerConfig config() {
Properties prop = new Properties();
prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class);
@@ -1786,4 +1843,23 @@ class StreamsGroupHeartbeatRequestManagerTest {
.collect(Collectors.toList());
assertEquals(sortedExpected, sortedActual);
}
+
+ private ClientResponse buildClientResponseWithTopologyRequired(final
boolean topologyRequired) {
+ return new ClientResponse(
+ new RequestHeader(ApiKeys.STREAMS_GROUP_HEARTBEAT, (short) 1, "",
1),
+ null,
+ "-1",
+ time.milliseconds(),
+ time.milliseconds(),
+ false,
+ null,
+ null,
+ new StreamsGroupHeartbeatResponse(
+ new StreamsGroupHeartbeatResponseData()
+ .setPartitionsByUserEndpoint(ENDPOINT_TO_PARTITIONS)
+ .setHeartbeatIntervalMs((int)
RECEIVED_HEARTBEAT_INTERVAL_MS)
+ .setTopologyDescriptionRequired(topologyRequired)
+ )
+ );
+ }
}
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupTopologyDescriptionRequestManagerTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupTopologyDescriptionRequestManagerTest.java
new file mode 100644
index 00000000000..05eb5d032a3
--- /dev/null
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupTopologyDescriptionRequestManagerTest.java
@@ -0,0 +1,342 @@
+/*
+ * 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.clients.consumer.internals;
+
+import org.apache.kafka.clients.ClientResponse;
+import org.apache.kafka.common.Node;
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
+import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.RequestHeader;
+import
org.apache.kafka.common.requests.StreamsGroupTopologyDescriptionUpdateRequest;
+import
org.apache.kafka.common.requests.StreamsGroupTopologyDescriptionUpdateResponse;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.internals.LogContext;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+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.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class StreamsGroupTopologyDescriptionRequestManagerTest {
+ private static final String GROUP_ID = "group-id";
+ private static final String MEMBER_ID = "member-id";
+ private static final long RETRY_BACKOFF_MS = 100;
+ private static final long RETRY_BACKOFF_MAX_MS = 1000;
+
+ private final MockTime time = new MockTime();
+ private final LogContext logContext = new LogContext();
+ private final Node coordinatorNode = new Node(1, "host", 9092);
+
+ private CoordinatorRequestManager coordinatorRequestManager;
+ private StreamsMembershipManager membershipManager;
+ private StreamsRebalanceData streamsRebalanceData;
+ private StreamsGroupTopologyDescriptionRequestManager manager;
+
+ @BeforeEach
+ public void setUp() {
+ coordinatorRequestManager = mock(CoordinatorRequestManager.class);
+ membershipManager = mock(StreamsMembershipManager.class);
+ when(membershipManager.groupId()).thenReturn(GROUP_ID);
+ when(membershipManager.memberId()).thenReturn(MEMBER_ID);
+ streamsRebalanceData = new StreamsRebalanceData(
+ UUID.randomUUID(), Optional.empty(), Optional.empty(), Map.of(),
Map.of(), Map::of, Map::of
+ );
+ manager = new StreamsGroupTopologyDescriptionRequestManager(
+ logContext, time, RETRY_BACKOFF_MS, RETRY_BACKOFF_MAX_MS,
+ membershipManager, streamsRebalanceData, coordinatorRequestManager
+ );
+ }
+
+ /**
+ * Test the situation when the coordinator has not yet been discovered: no
request should be sent
+ * and the client should silently wait for the coordinator to become known.
+ */
+ @Test
+ public void testCoordinatorUnknown() {
+ streamsRebalanceData.setWireTopologyDescription(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription());
+ streamsRebalanceData.setTopologyPushRequired(true);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.empty());
+
+ assertEquals(0,
manager.poll(time.milliseconds()).unsentRequests.size());
+ }
+
+ /**
+ * Test the situation when the wire-format topology description has not
yet been populated:
+ * no request should be sent.
+ */
+ @Test
+ public void testWireTopologyDescriptionNull() {
+ streamsRebalanceData.setTopologyPushRequired(true);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(coordinatorNode));
+
+ assertEquals(0,
manager.poll(time.milliseconds()).unsentRequests.size());
+ }
+
+ /**
+ * Test the situation when memberId is null because the member has not yet
joined the group:
+ * no request should be sent.
+ */
+ @Test
+ public void testMemberIdNull() {
+ streamsRebalanceData.setWireTopologyDescription(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription());
+ streamsRebalanceData.setTopologyPushRequired(true);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(coordinatorNode));
+ when(membershipManager.memberId()).thenReturn(null);
+
+ assertEquals(0,
manager.poll(time.milliseconds()).unsentRequests.size());
+ }
+
+ /**
+ * Test the situation when the topologyPushRequired flag is not set:
maximumTimeToWait should
+ * return Long.MAX_VALUE and no request should be sent even if every other
precondition is satisfied.
+ */
+ @Test
+ public void testTopologyDescriptionWhenPushFlagNotSet() {
+ streamsRebalanceData.setWireTopologyDescription(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription());
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(coordinatorNode));
+
+ assertEquals(Long.MAX_VALUE,
manager.maximumTimeToWait(time.milliseconds()));
+ assertEquals(0,
manager.poll(time.milliseconds()).unsentRequests.size());
+ }
+
+ /**
+ * Test the situation when all preconditions are satisfied:
maximumTimeToWait should return 0
+ * to wake the poll loop immediately, and the subsequent poll should build
a request with the
+ * correct groupId, memberId, and target coordinator node.
+ */
+ @Test
+ public void testSendRequestWhenAllPreconditionsMet() {
+ streamsRebalanceData.setWireTopologyDescription(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription());
+ streamsRebalanceData.setTopologyPushRequired(true);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(coordinatorNode));
+
+ assertEquals(0L, manager.maximumTimeToWait(time.milliseconds()));
+
+ final NetworkClientDelegate.PollResult result =
manager.poll(time.milliseconds());
+
+ assertEquals(1, result.unsentRequests.size());
+ final NetworkClientDelegate.UnsentRequest unsent =
result.unsentRequests.get(0);
+ assertEquals(Optional.of(coordinatorNode), unsent.node());
+ final StreamsGroupTopologyDescriptionUpdateRequest request =
+ (StreamsGroupTopologyDescriptionUpdateRequest)
unsent.requestBuilder().build();
+ assertEquals(GROUP_ID, request.data().groupId());
+ assertEquals(MEMBER_ID, request.data().memberId());
+ }
+
+ /**
+ * Test the situation when a push request is already in flight: subsequent
polls should not
+ * enqueue duplicate requests.
+ */
+ @Test
+ public void testNoSecondRequestWhileInflight() {
+ streamsRebalanceData.setWireTopologyDescription(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription());
+ streamsRebalanceData.setTopologyPushRequired(true);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(coordinatorNode));
+
+ manager.poll(time.milliseconds());
+ assertEquals(0,
manager.poll(time.milliseconds()).unsentRequests.size());
+ }
+
+ /**
+ * Test the situation when the broker responds with no error: the
topologyPushRequired flag
+ * should be cleared so the manager stops pushing.
+ */
+ @Test
+ public void testFlagClearedOnSuccess() {
+ streamsRebalanceData.setWireTopologyDescription(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription());
+ streamsRebalanceData.setTopologyPushRequired(true);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(coordinatorNode));
+
+ final NetworkClientDelegate.UnsentRequest unsent =
+ manager.poll(time.milliseconds()).unsentRequests.get(0);
+ unsent.handler().onComplete(buildResponse(Errors.NONE, 0));
+
+ assertFalse(streamsRebalanceData.topologyPushRequired());
+ }
+
+ /**
+ * Test the situation when the broker responds with UNKNOWN_MEMBER_ID: the
flag should be
+ * cleared so the heartbeat path can drive a clean rejoin.
+ */
+ @Test
+ public void testFlagClearedOnUnknownMemberId() {
+ streamsRebalanceData.setWireTopologyDescription(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription());
+ streamsRebalanceData.setTopologyPushRequired(true);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(coordinatorNode));
+
+ final NetworkClientDelegate.UnsentRequest unsent =
+ manager.poll(time.milliseconds()).unsentRequests.get(0);
+ unsent.handler().onComplete(buildResponse(Errors.UNKNOWN_MEMBER_ID,
0));
+
+ assertFalse(streamsRebalanceData.topologyPushRequired());
+ }
+
+ /**
+ * Test the situation when the broker responds with NOT_COORDINATOR or
COORDINATOR_NOT_AVAILABLE:
+ * coordinator rediscovery should be triggered and the flag should remain
set for retry.
+ */
+ @Test
+ public void testCoordinatorErrorTriggersRediscovery() {
+ for (final Errors error : new Errors[]{Errors.NOT_COORDINATOR,
Errors.COORDINATOR_NOT_AVAILABLE}) {
+ streamsRebalanceData.setWireTopologyDescription(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription());
+ streamsRebalanceData.setTopologyPushRequired(true);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(coordinatorNode));
+ time.sleep(RETRY_BACKOFF_MAX_MS);
+
+ final NetworkClientDelegate.UnsentRequest unsent =
+ manager.poll(time.milliseconds()).unsentRequests.get(0);
+ unsent.handler().onComplete(buildResponse(error, 0));
+
+ assertTrue(streamsRebalanceData.topologyPushRequired(),
+ "Flag should remain set after " + error);
+ }
+ verify(coordinatorRequestManager,
times(2)).markCoordinatorUnknown(any(), anyLong());
+ }
+
+ /**
+ * Test the situation when the broker responds with
COORDINATOR_LOAD_IN_PROGRESS: the flag
+ * should remain set for retry but coordinator rediscovery should not be
triggered, since the
+ * coordinator is correct but still loading.
+ */
+ @Test
+ public void testCoordinatorLoadInProgressDoesNotTriggerRediscovery() {
+ streamsRebalanceData.setWireTopologyDescription(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription());
+ streamsRebalanceData.setTopologyPushRequired(true);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(coordinatorNode));
+
+ final NetworkClientDelegate.UnsentRequest unsent =
+ manager.poll(time.milliseconds()).unsentRequests.get(0);
+
unsent.handler().onComplete(buildResponse(Errors.COORDINATOR_LOAD_IN_PROGRESS,
0));
+
+ assertTrue(streamsRebalanceData.topologyPushRequired());
+ verify(coordinatorRequestManager,
never()).markCoordinatorUnknown(any(), anyLong());
+ }
+
+ /**
+ * Test the situation when the broker responds with a terminal error
listed in KIP-1331: the
+ * flag should be cleared and the client should not retry on its own.
+ */
+ @Test
+ public void testTerminalErrorsClearFlag() {
+ for (final Errors error : new Errors[]{
+ Errors.STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED,
+ Errors.INVALID_REQUEST,
+ Errors.UNSUPPORTED_VERSION,
+ Errors.GROUP_ID_NOT_FOUND,
+ Errors.GROUP_AUTHORIZATION_FAILED}) {
+ streamsRebalanceData.setWireTopologyDescription(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription());
+ streamsRebalanceData.setTopologyPushRequired(true);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(coordinatorNode));
+ time.sleep(RETRY_BACKOFF_MAX_MS);
+
+ final NetworkClientDelegate.UnsentRequest unsent =
+ manager.poll(time.milliseconds()).unsentRequests.get(0);
+ unsent.handler().onComplete(buildResponse(error, 0));
+
+ assertFalse(streamsRebalanceData.topologyPushRequired(),
+ "Flag should be cleared after " + error);
+ }
+ }
+
+ /**
+ * Test the situation when the push fails with a network exception: the
+ * topologyPushRequired flag must remain set so the next poll retries, and
a failed attempt
+ * is recorded so retry backoff applies.
+ */
+ @Test
+ public void testNetworkExceptionLeavesFlagSetWithBackoff() {
+ streamsRebalanceData.setWireTopologyDescription(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription());
+ streamsRebalanceData.setTopologyPushRequired(true);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(coordinatorNode));
+
+ final NetworkClientDelegate.UnsentRequest unsent =
+ manager.poll(time.milliseconds()).unsentRequests.get(0);
+ unsent.handler().onFailure(time.milliseconds(),
Errors.NETWORK_EXCEPTION.exception());
+
+ assertTrue(streamsRebalanceData.topologyPushRequired());
+ assertEquals(0,
manager.poll(time.milliseconds()).unsentRequests.size());
+
+ time.sleep(RETRY_BACKOFF_MAX_MS);
+ assertEquals(1,
manager.poll(time.milliseconds()).unsentRequests.size());
+ }
+
+ /**
+ * Test the situation when the response carries a non-zero ThrottleTimeMs:
subsequent push
+ * attempts should be delayed by that amount before resuming.
+ */
+ @Test
+ public void testThrottleDelayBlocksSubsequentRequest() {
+ streamsRebalanceData.setWireTopologyDescription(
+ new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription());
+ streamsRebalanceData.setTopologyPushRequired(true);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(coordinatorNode));
+
+ final NetworkClientDelegate.UnsentRequest unsent =
+ manager.poll(time.milliseconds()).unsentRequests.get(0);
+ unsent.handler().onComplete(buildResponse(Errors.NOT_COORDINATOR,
500));
+
+ streamsRebalanceData.setTopologyPushRequired(true);
+ assertEquals(0,
manager.poll(time.milliseconds()).unsentRequests.size());
+
+ time.sleep(501);
+ assertEquals(1,
manager.poll(time.milliseconds()).unsentRequests.size());
+ }
+
+ private ClientResponse buildResponse(final Errors error, final int
throttleTimeMs) {
+ return new ClientResponse(
+ new
RequestHeader(ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE, (short) 0, "",
1),
+ null,
+ "-1",
+ time.milliseconds(),
+ time.milliseconds(),
+ false,
+ null,
+ null,
+ new StreamsGroupTopologyDescriptionUpdateResponse(
+ new StreamsGroupTopologyDescriptionUpdateResponseData()
+ .setErrorCode(error.code())
+ .setThrottleTimeMs(throttleTimeMs)
+ )
+ );
+ }
+}
diff --git
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
index 60403008a8b..1f56728ae11 100644
---
a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
@@ -35,6 +35,7 @@ import
org.apache.kafka.clients.consumer.internals.ShareConsumeRequestManager;
import
org.apache.kafka.clients.consumer.internals.ShareHeartbeatRequestManager;
import org.apache.kafka.clients.consumer.internals.ShareMembershipManager;
import
org.apache.kafka.clients.consumer.internals.StreamsGroupHeartbeatRequestManager;
+import
org.apache.kafka.clients.consumer.internals.StreamsGroupTopologyDescriptionRequestManager;
import org.apache.kafka.clients.consumer.internals.StreamsMembershipManager;
import org.apache.kafka.clients.consumer.internals.SubscriptionState;
import org.apache.kafka.clients.consumer.internals.TopicMetadataRequestManager;
@@ -97,6 +98,7 @@ public class ApplicationEventProcessorTest {
private SubscriptionState subscriptionState =
mock(SubscriptionState.class);
private final ConsumerMetadata metadata = mock(ConsumerMetadata.class);
private final StreamsGroupHeartbeatRequestManager
streamsGroupHeartbeatRequestManager =
mock(StreamsGroupHeartbeatRequestManager.class);
+ private final StreamsGroupTopologyDescriptionRequestManager
streamsGroupTopologyDescriptionRequestManager =
mock(StreamsGroupTopologyDescriptionRequestManager.class);
private final StreamsMembershipManager streamsMembershipManager =
mock(StreamsMembershipManager.class);
private final ShareHeartbeatRequestManager shareHeartbeatRequestManager =
mock(ShareHeartbeatRequestManager.class);
private final ShareMembershipManager shareMembershipManager =
mock(ShareMembershipManager.class);
@@ -113,6 +115,7 @@ public class ApplicationEventProcessorTest {
withGroupId ? Optional.of(heartbeatRequestManager) :
Optional.empty(),
withGroupId ? Optional.of(membershipManager) :
Optional.empty(),
Optional.empty(),
+ Optional.empty(),
Optional.empty()
);
processor = new ApplicationEventProcessor(
@@ -134,6 +137,7 @@ public class ApplicationEventProcessorTest {
withGroupId ? Optional.of(heartbeatRequestManager) :
Optional.empty(),
Optional.empty(),
withGroupId ? Optional.of(streamsGroupHeartbeatRequestManager) :
Optional.empty(),
+ withGroupId ?
Optional.of(streamsGroupTopologyDescriptionRequestManager) : Optional.empty(),
withGroupId ? Optional.of(streamsMembershipManager) :
Optional.empty()
);
processor = new ApplicationEventProcessor(
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
index 9084e7f8d5f..bb85da626b0 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
@@ -700,7 +700,8 @@ public class StreamThread extends Thread implements
ProcessingThread {
return Optional.of(rackId);
}
- private static StreamsRebalanceData initStreamsRebalanceData(final UUID
processId,
+ // visible for testing
+ static StreamsRebalanceData initStreamsRebalanceData(final UUID processId,
final
StreamsConfig config,
final
Optional<StreamsRebalanceData.HostInfo> endpoint,
final
Optional<String> rackId,
@@ -1043,7 +1044,7 @@ public class StreamThread extends Thread implements
ProcessingThread {
return true;
}
- // visible for testing
+ // VisibleForTesting
void maybeGetClientInstanceIds() {
// we pass in a timeout of zero into each `clientInstanceId()` call
// to just trigger the "get instance id" background RPC;
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
index a6db72daf56..2c0223fc420 100644
---
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
@@ -4514,6 +4514,52 @@ public class StreamThreadTest {
);
}
+ @Test
+ public void shouldPopulateWireTopologyDescriptionWhenPushEnabled() {
+ internalTopologyBuilder.addSource(null, "source1", null, null, null,
"input-topic");
+ internalTopologyBuilder.addSink("sink1", "output-topic", null, null,
null, "source1");
+
+ final StreamsConfig config = new StreamsConfig(configProps(false,
false));
+ final TopologyMetadata topologyMetadata = new
TopologyMetadata(internalTopologyBuilder, config);
+
+ final StreamsRebalanceData rebalanceData =
StreamThread.initStreamsRebalanceData(
+ UUID.randomUUID(),
+ config,
+ Optional.empty(),
+ Optional.empty(),
+ topologyMetadata,
+ Map::of,
+ Map::of
+ );
+
+ assertNotNull(rebalanceData.wireTopologyDescription());
+
assertFalse(rebalanceData.wireTopologyDescription().subtopologies().isEmpty());
+ }
+
+ @Test
+ public void shouldNotPopulateWireTopologyDescriptionWhenPushDisabled() {
+ internalTopologyBuilder.addSource(null, "source1", null, null, null,
"input-topic");
+ internalTopologyBuilder.addSink("sink1", "output-topic", null, null,
null, "source1");
+
+ final Properties props = configProps(false, false);
+
props.setProperty(StreamsConfig.TOPOLOGY_DESCRIPTION_PUSH_ENABLED_CONFIG,
"false");
+ final StreamsConfig config = new StreamsConfig(props);
+ final TopologyMetadata topologyMetadata = new
TopologyMetadata(internalTopologyBuilder, config);
+
+ final StreamsRebalanceData rebalanceData =
StreamThread.initStreamsRebalanceData(
+ UUID.randomUUID(),
+ config,
+ Optional.empty(),
+ Optional.empty(),
+ topologyMetadata,
+ Map::of,
+ Map::of
+ );
+
+ assertNull(rebalanceData.wireTopologyDescription());
+ }
+
+
private StreamThread setUpThread(final Properties streamsConfigProps) {
final StreamsConfig config = new StreamsConfig(streamsConfigProps);
final ConsumerGroupMetadata consumerGroupMetadata =
Mockito.mock(ConsumerGroupMetadata.class);
