lucasbru commented on code in PR #23761:
URL: https://github.com/apache/kafka/pull/23761#discussion_r4246317615
##########
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 is already reworded in 4d188e7261: the javadoc now states the contract
for implementations (must not newly place a warm-up task on a member that is
not supported) and that the coordinator reconciles towards the target
assignment if it does.
##########
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:
A cached intermediate assignment was validated when it was derived, and I
don't see a path where a member's support changes afterwards. A member ID
belongs to one client instance, and a restart gets a new member ID, so an
UNKNOWN member can only hold a cached warm-up if it is a version 1 client that
held it before the failover. Revalidating on every heartbeat would also add an
O(members) pass to each heartbeat of a reconciling group. Happy to add it if
you see a way a member's version can change under the same member ID.
##########
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:
Fixed in 82dfbcc351 (that was also what failed spotless in CI).
##########
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:
The refiner is handed `group.members()` as well, which is the group state
without a member whose join record has not been replayed yet, so
`warmupSupport()` and the refiner see the same membership. A joining member is
not in the refiner's input either, and its target tasks are parked on their old
owners until it is part of the group. So no warm-up can be placed on it by this
derivation, and none is discarded.
--
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]
