cadonna commented on code in PR #19181:
URL: https://github.com/apache/kafka/pull/19181#discussion_r1991827905
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -82,29 +97,41 @@ public StreamsGroupHeartbeatRequestData buildRequestData() {
data.setMemberId(membershipManager.memberId());
data.setMemberEpoch(membershipManager.memberEpoch());
membershipManager.groupInstanceId().ifPresent(data::setInstanceId);
- StreamsGroupHeartbeatRequestData.Topology topology = new
StreamsGroupHeartbeatRequestData.Topology();
-
topology.setSubtopologies(getTopologyFromStreams(streamsRebalanceData.subtopologies()));
- topology.setEpoch(streamsRebalanceData.topologyEpoch());
- data.setRebalanceTimeoutMs(rebalanceTimeoutMs);
- data.setTopology(topology);
- data.setProcessId(streamsRebalanceData.processId().toString());
- streamsRebalanceData.endpoint().ifPresent(userEndpoint -> {
- data.setUserEndpoint(new
StreamsGroupHeartbeatRequestData.Endpoint()
- .setHost(userEndpoint.host())
- .setPort(userEndpoint.port())
- );
- });
-
data.setClientTags(streamsRebalanceData.clientTags().entrySet().stream()
- .map(entry -> new StreamsGroupHeartbeatRequestData.KeyValue()
- .setKey(entry.getKey())
- .setValue(entry.getValue())
- )
- .collect(Collectors.toList()));
+
+ boolean joining = membershipManager.state() == MemberState.JOINING;
+
+ if (joining) {
+ StreamsGroupHeartbeatRequestData.Topology topology = new
StreamsGroupHeartbeatRequestData.Topology();
+
topology.setSubtopologies(getTopologyFromStreams(streamsRebalanceData.subtopologies()));
+ topology.setEpoch(streamsRebalanceData.topologyEpoch());
+ data.setTopology(topology);
+ data.setRebalanceTimeoutMs(rebalanceTimeoutMs);
+ data.setProcessId(streamsRebalanceData.processId().toString());
+ streamsRebalanceData.endpoint().ifPresent(userEndpoint -> {
+ data.setUserEndpoint(new
StreamsGroupHeartbeatRequestData.Endpoint()
+ .setHost(userEndpoint.host())
+ .setPort(userEndpoint.port())
+ );
+ });
+
data.setClientTags(streamsRebalanceData.clientTags().entrySet().stream()
+ .map(entry -> new
StreamsGroupHeartbeatRequestData.KeyValue()
+ .setKey(entry.getKey())
+ .setValue(entry.getValue())
+ )
+ .collect(Collectors.toList()));
+ data.setActiveTasks(convertTaskIdCollection(Set.of()));
Review Comment:
The protocol specifies that the assignment needs to consists of empty sets
when joining. The heartbeat handler broker side distinguishes between an empty
assignment and an assignment that was not sent in the heartbeat.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
