gabriellefu commented on code in PR #23165:
URL: https://github.com/apache/kafka/pull/23165#discussion_r3833248817
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImpl.java:
##########
@@ -52,20 +53,16 @@ public record AssignmentConfigsImpl(
}
/**
- * Converts the raw assignment configs computed for the group into the
typed configs passed to the assignor.
+ * Converts the raw assignment configs recorded for the group into the
typed configs passed to the assignor.
*/
public static AssignmentConfigsImpl fromMap(Map<String, String> configs) {
- // The map is empty when it was replayed from a group metadata record
written before the last assignment
- // configs were persisted.
- if (configs.isEmpty()) {
- return DEFAULT;
- }
- // The rack-aware assignment tags are only recorded when any are
configured, so an absent value means the
- // configuration is at its default.
- String rackAwareAssignmentTags = configs.getOrDefault(
- RACK_AWARE_ASSIGNMENT_TAGS_CONFIG,
GroupCoordinatorConfig.STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT);
+ // Configs are only recorded when set, so every absent key means the
configuration is at its default.
+ String numStandbyReplicas =
configs.getOrDefault(NUM_STANDBY_REPLICAS_CONFIG,
+
Integer.toString(GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT));
+ String rackAwareAssignmentTags =
configs.getOrDefault(RACK_AWARE_ASSIGNMENT_TAGS_CONFIG,
+
GroupCoordinatorConfig.STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT);
return new AssignmentConfigsImpl(
- Integer.parseInt(configs.get(NUM_STANDBY_REPLICAS_CONFIG)),
+ Integer.parseInt(numStandbyReplicas),
Review Comment:
got it, thanks
--
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]
