lucasbru commented on code in PR #19114:
URL: https://github.com/apache/kafka/pull/19114#discussion_r1987320429
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1805,6 +1990,277 @@ private
List<ShareGroupHeartbeatResponseData.TopicPartitions> fromShareGroupAssi
.toList();
}
+ /**
+ * Handles a regular heartbeat from a Streams group member.
+ * It mainly consists of five parts:
+ * 1) Created or update the member.
+ * The group epoch is bumped if the member has been created or updated.
+ * 2) Initialized or update the topology. The group epoch is bumped if the
topology
+ * has been created or updated.
+ * 3) Determine the partition metadata and any internal topics that need
to be created.
+ * 4) Update the target assignment for the streams group if the group epoch
+ * is larger than the current target assignment epoch.
+ * 5) Reconcile the member's assignment with the target assignment.
+ *
+ * @param groupId The group id from the request.
+ * @param memberId The member ID from the request.
+ * @param memberEpoch The member epoch from the request.
+ * @param instanceId The instance ID from the request or null.
+ * @param rackId The rack ID from the request or null.
+ * @param rebalanceTimeoutMs The rebalance timeout from the request or -1.
+ * @param clientId The client ID.
+ * @param clientHost The client host.
+ * @param topology The topology from the request or null.
+ * @param ownedActiveTasks The list of owned active tasks from the
request or null.
+ * @param ownedStandbyTasks The list of owned standby tasks from the
request or null.
+ * @param ownedWarmupTasks The list of owned warmup tasks from the
request or null.
+ * @param userEndpoint User-defined endpoint for Interactive
Queries, or null.
+ * @param clientTags Used for rack-aware assignment algorithm, or
null.
+ * @param taskEndOffsets Cumulative changelog offsets for tasks, or
null.
+ * @param taskOffsets Cumulative changelog end-offsets for tasks,
or null.
+ * @param shutdownApplication Whether all Streams clients in the group
should shut down.
+ * @return A Result containing the StreamsGroupHeartbeat response and a
list of records to update the state machine.
+ */
+ private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord>
streamsGroupHeartbeat(
+ String groupId,
+ String memberId,
+ int memberEpoch,
+ String instanceId,
+ String rackId,
+ int rebalanceTimeoutMs,
+ String clientId,
+ String clientHost,
+ final StreamsGroupHeartbeatRequestData.Topology topology,
+ List<TaskIds> ownedActiveTasks,
+ List<TaskIds> ownedStandbyTasks,
+ List<TaskIds> ownedWarmupTasks,
+ final String processId,
+ final Endpoint userEndpoint,
+ final List<KeyValue> clientTags,
+ final List<TaskOffset> taskOffsets,
+ final List<TaskOffset> taskEndOffsets,
+ final boolean shutdownApplication
+ ) throws ApiException {
+ final long currentTimeMs = time.milliseconds();
+ final List<CoordinatorRecord> records = new ArrayList<>();
+ final List<StreamsGroupHeartbeatResponseData.Status> returnedStatus =
new ArrayList<>();
+
+ // Get or create the streams group.
+ boolean createIfNotExists = memberEpoch == 0;
+ final StreamsGroup group = getOrMaybeCreateStreamsGroup(groupId,
createIfNotExists);
+
+ // Get or create the member.
+ StreamsGroupMember member;
+ if (instanceId == null) {
+ member = getOrMaybeSubscribeDynamicStreamsGroupMember(
+ group,
+ memberId,
+ memberEpoch,
+ ownedActiveTasks,
+ ownedStandbyTasks,
+ ownedWarmupTasks,
+ createIfNotExists
+ );
+ } else {
+ throw new UnsupportedOperationException("Static members are not
supported yet.");
+ }
+
+ // 1. Create or update the member.
+ // If the member is new or has changed, a StreamsMemberMetadataValue
record is written to the __consumer_offsets partition to
+ // persist the change. If the subscriptions have changed, the
subscription metadata is updated and persisted by writing a
+ // StreamsPartitionMetadataValue
+ // record to the __consumer_offsets partition. Finally, the group
epoch is bumped if the subscriptions have
+ // changed, and persisted by writing a StreamsMetadataValue record to
the partition.
+ StreamsGroupMember updatedMember = new
StreamsGroupMember.Builder(member)
+ .maybeUpdateInstanceId(Optional.empty())
+ .maybeUpdateRackId(Optional.ofNullable(rackId))
+ .maybeUpdateRebalanceTimeoutMs(ofSentinel(rebalanceTimeoutMs))
+ .maybeUpdateTopologyEpoch(topology != null ?
OptionalInt.of(topology.epoch()) : OptionalInt.empty())
+ .setClientId(clientId)
+ .setClientHost(clientHost)
+ .maybeUpdateProcessId(Optional.ofNullable(processId))
+ .maybeUpdateClientTags(Optional.ofNullable(clientTags).map(x ->
x.stream().collect(Collectors.toMap(KeyValue::key, KeyValue::value))))
+ .maybeUpdateUserEndpoint(Optional.ofNullable(userEndpoint).map(x
-> new
StreamsGroupMemberMetadataValue.Endpoint().setHost(x.host()).setPort(x.port())))
+ .build();
+
+ // If the group is newly created, we must ensure that it moves away
from
+ // epoch 0 and that it is fully initialized.
+ int groupEpoch = group.groupEpoch();
+ boolean bumpGroupEpoch = false;
+
+ bumpGroupEpoch |= hasStreamsMemberMetadataChanged(groupId, member,
updatedMember, records);
+
+ // 2. Initialize/Update the group topology.
+ // If the member is new or has changed, a StreamsGroupTopologyValue
record is written to the __consumer_offsets partition to persist
+ // the change. The group epoch is bumped if the topology has changed.
+ StreamsTopology updatedTopology;
+ boolean reconfigureTopology = false;
+ if (topology != null) {
+ StreamsGroupTopologyValue recordValue =
convertToStreamsGroupTopologyRecord(topology);
+
+ updatedTopology = StreamsTopology.fromRequest(topology);
Review Comment:
Done
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -355,6 +355,32 @@ public StreamsGroupMember getOrMaybeCreateMember(
return new StreamsGroupMember.Builder(memberId).build();
}
+ /**
+ * Gets or creates a new member but without adding it to the group. Adding
a member is done via the
+ * {@link StreamsGroup#updateMember(StreamsGroupMember)} method.
+ *
+ * @param memberId The member ID.
+ * @param createIfNotExists Booleans indicating whether the member must be
created if it does not exist.
+ * @return A StreamsGroupMember.
+ */
+ public StreamsGroupMember getOrMaybeCreateDefaultMember(
Review Comment:
Done
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
