Copilot commented on code in PR #23761:
URL: https://github.com/apache/kafka/pull/23761#discussion_r4230099843
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -4718,6 +4724,19 @@ private Map<String, TasksTuple> refineAssignment(
group.groupId(), targetAssignment, refinedAssignment);
return targetAssignment;
}
+ final Set<String> unsupportedWarmupHolders =
+
AssignmentRefiner.membersWithUnsupportedWarmupTasks(refinedAssignment,
warmupSupport, group.members());
Review Comment:
This validation is bypassed when an intermediate assignment is already
cached: `refinedAssignmentForEpoch` returns the cached value at lines 4761–4764
without calling this method. After a failover, an UNKNOWN member may
legitimately keep a cached warm-up; if that member then heartbeats with version
0 while the group is still RECONCILING, the coordinator reuses the cache and
continues reconciling the unsupported member toward that warm-up instead of
falling back to the target assignment. Revalidate cached assignments against
the current support map (and replace/evict the cache on failure), or invalidate
the cache when support becomes NOT_SUPPORTED.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -602,6 +612,29 @@ public void updateTaskOffsets(String memberId,
MemberTaskOffsets memberOffsets)
taskOffsets.put(memberId, memberOffsets);
}
+ /**
+ * Records the version of the latest streams group heartbeat of a member.
This is transient and not persisted.
+ *
+ * @param memberId The member ID.
+ * @param heartbeatVersion The version of the heartbeat request.
+ */
+ public void updateHeartbeatVersion(String memberId, int heartbeatVersion) {
+ heartbeatVersions.put(memberId, heartbeatVersion);
+ }
+
+ /**
+ * @return Whether each member of the group may be handed a warm-up task,
keyed by member ID. A member whose
+ * heartbeat version is not known is {@link WarmupSupport#UNKNOWN}.
+ */
+ public Map<String, WarmupSupport> warmupSupport() {
+ final Map<String, WarmupSupport> warmupSupport = new HashMap<>();
+ members.keySet().forEach(memberId -> {
+ final Integer version = heartbeatVersions.get(memberId);
+ warmupSupport.put(memberId, version == null ?
WarmupSupport.UNKNOWN : WarmupSupport.ofHeartbeatVersion(version));
Review Comment:
This only emits support for members already replayed into `members`, but the
heartbeat path calls `updateHeartbeatVersion` before a joining member's
metadata record is replayed. During a join or static-member replacement, the
just-observed v1 member is therefore omitted and treated as UNKNOWN by both the
refiner contract and the guard, causing a valid warm-up placement for that
newcomer to be discarded. Include recorded-but-not-yet-replayed members, or
derive support from the same pending membership view used for target assignment.
##########
jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/AssignmentRefinerBenchmark.java:
##########
@@ -24,6 +24,7 @@
import org.apache.kafka.coordinator.group.streams.NoOpAssignmentRefiner;
import org.apache.kafka.coordinator.group.streams.StreamsGroupMember;
import org.apache.kafka.coordinator.group.streams.TasksTuple;
+import org.apache.kafka.coordinator.group.streams.WarmupSupport;
import org.apache.kafka.coordinator.group.streams.TopologyMetadata;
Review Comment:
Imports in this group are lexicographically ordered (as the adjacent
`TasksTuple` and `TopologyMetadata` imports show), but `WarmupSupport` is
inserted before `TopologyMetadata`. Swap these imports to satisfy the enforced
import-order formatting.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefiner.java:
##########
@@ -69,6 +72,10 @@ public interface AssignmentRefiner {
* The latest changelog offsets/end-offsets reported by the members
via heartbeats, from which the lag
* of a warm-up task is derived. Not populated for a member that
has not reported any offsets yet, for
* example right after a coordinator failover.
+ * @param warmupSupport
+ * Whether each member may be handed a warm-up task, keyed by
member ID. A member that is missing from the
+ * map is {@link WarmupSupport#UNKNOWN}. The refiner never places a
warm-up task on a member that is not
+ * {@link WarmupSupport#SUPPORTED}, except that it keeps one that a
member of unknown version already holds.
Review Comment:
This states that the refiner enforces the restriction, but
`AssignmentRefinerImpl` intentionally does not use this parameter in this
change; enforcement currently happens in the coordinator after `refine`
returns. Document the coordinator-side validation instead so implementers do
not rely on a guarantee the built-in implementation does not provide.
--
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]
