lucasbru commented on code in PR #23761:
URL: https://github.com/apache/kafka/pull/23761#discussion_r4246548129
##########
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));
+ });
+ return warmupSupport;
Review Comment:
Fixed in 7368ea6f20: `warmupSupport()` returns an unmodifiable snapshot,
with a test that it cannot be modified.
--
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]
