You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mjsax (via GitHub)" <gi...@apache.org> on 2026/06/04 06:22:48 UTC
Re: [PR] KSTREAMS-8781: Add back `rack.aware.assignment.tags` config [kafka]
mjsax commented on code in PR #22213:
URL: https://github.com/apache/kafka/pull/22213#discussion_r3353818484
##########
clients/src/main/resources/common/message/StreamsGroupHeartbeatRequest.json:
##########
@@ -18,7 +18,7 @@
"type": "request",
"listeners": ["broker"],
"name": "StreamsGroupHeartbeatRequest",
- "validVersions": "0",
+ "validVersions": "0-1",
Review Comment:
The version was bumped in some other PR in the mean time, so we don't need to do this any longer. That's also why there are merge conflicts. Rebasing to `trunk` will fix this issues.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -541,10 +543,21 @@ private void onSuccessResponse(final StreamsGroupHeartbeatResponse response, fin
if (statuses != null) {
streamsRebalanceData.setStatuses(statuses);
if (!statuses.isEmpty()) {
+ for (StreamsGroupHeartbeatResponseData.Status status : statuses) {
Review Comment:
Seems we are looping over `statuses` twice now. In this new for-loop and in the below `statuses.streams()...` code.
Might be good to unify the code to only loop onec?
##########
clients/src/main/resources/common/message/StreamsGroupHeartbeatResponse.json:
##########
@@ -17,7 +17,7 @@
"apiKey": 88,
"type": "response",
"name": "StreamsGroupHeartbeatResponse",
- "validVersions": "0",
+ "validVersions": "0-1",
Review Comment:
same
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -693,7 +699,6 @@ public GroupCoordinatorConfig(AbstractConfig config) {
String.format("%s must be less than or equal to %s",
SHARE_GROUP_ASSIGNMENT_INTERVAL_MS_CONFIG, SHARE_GROUP_MAX_ASSIGNMENT_INTERVAL_MS_CONFIG));
-
Review Comment:
I see -- well, instead of doing this, let's split out a helper method `verifyConfigs()` (or maybe even multiple, one for each "group of configs", like `verifyShareGroupConfigs()`, `verifyStreamsGroupConfigs()` and similar), to make the method really shorter.
##########
clients/src/main/resources/common/message/StreamsGroupHeartbeatResponse.json:
##########
@@ -52,7 +52,6 @@
"about": "The maximal lag a warm-up task can have to be considered caught-up." },
{ "name": "TaskOffsetIntervalMs", "type": "int32", "versions": "0+",
"about": "The interval in which the task changelog offsets on a client are updated on the broker. The offsets are sent with the next heartbeat after this time has passed." },
-
Review Comment:
nit: avoid unnecessary reformatting
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2189,6 +2190,27 @@ private CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord> stream
)
));
+ String rackAwareTagsValue = currentAssignmentConfigs.getOrDefault("rack.aware.assignment.tags", "");
+ if (!rackAwareTagsValue.isEmpty()) {
+ List<String> requiredTags = Arrays.stream(rackAwareTagsValue.split(",", -1))
+ .filter(tag -> !tag.isEmpty())
+ .collect(Collectors.toList());
+ Set<String> memberTagKeys = updatedMember.clientTags().keySet();
+ List<String> missingTags = requiredTags.stream()
+ .filter(tag -> !memberTagKeys.contains(tag))
+ .collect(Collectors.toList());
+ if (!missingTags.isEmpty()) {
+ returnedStatus.add(
+ new Status()
+ .setStatusCode(StreamsGroupHeartbeatResponse.Status.MISSING_CLIENT_TAGS.code())
+ .setStatusDetail(
+ String.format("Missing required client tags for rack-aware standby assignment: %s. " +
+ "Configure them via 'client.tag.<tagKey>' in your Streams config.", missingTags)
+ )
+ );
+ }
+ }
+
Review Comment:
Not sure. I am wondering if this is something the assignor needs to decide how to handle?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -8961,9 +8983,12 @@ private Map<String, String> streamsGroupAssignmentConfigs(String groupId) {
Optional<GroupConfig> groupConfig = groupConfigManager.groupConfig(groupId);
final Integer numStandbyReplicas = groupConfig.flatMap(GroupConfig::streamsNumStandbyReplicas)
.orElse(config.streamsGroupNumStandbyReplicas());
- return new TreeMap<>(Map.of(
- "num.standby.replicas", numStandbyReplicas.toString()
- ));
+ final List<String> rackAwareAssignmentTags = groupConfig.flatMap(GroupConfig::streamsRackAwareAssignmentTags)
+ .orElse(config.streamsGroupRackAwareAssignmentTags());
+ Map<String, String> configs = new TreeMap<>();
+ configs.put("num.standby.replicas", numStandbyReplicas.toString());
+ configs.put("rack.aware.assignment.tags", String.join(",", rackAwareAssignmentTags));
+ return configs;
Review Comment:
This is a valid point. Guess we have the same problem on https://github.com/apache/kafka/pull/21799
While we could maybe avoid it here (by not adding the config if it's the default, ie, empty string), not sure how to handle the case of PR 21799? And if we cannot fix it over there, it doesn't matter what we do here either.
Even if I am wondering about `acceptable.recovery.lag` now -- it is actually an "assignment configuration" we need to add here? Maybe we need to update PR 21799 to begin with, allowing us to change the code here to only add if non-empty string. \cc @lucasbru
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org