KAFKA-20790: Follow up, avoid translation of assignmentconfig and make the output type unmodifiable - #23165
KAFKA-20790: Follow up, avoid translation of assignmentconfig and make the output type unmodifiable#23165gabriellefu wants to merge 8 commits into
Conversation
| final int numStandbyReplicas = groupConfig.flatMap(GroupConfig::streamsNumStandbyReplicas) | ||
| .orElse(config.streamsGroupNumStandbyReplicas()); | ||
| final List<String> rackAwareAssignmentTags = groupConfig.flatMap(GroupConfig::streamsRackAwareAssignmentTags) | ||
| .orElse(config.streamsGroupRackAwareAssignmentTags()); |
There was a problem hiding this comment.
the GroupConfig::streamsRackAwareAssignmentTags and config.streamsGroupRackAwareAssignmentTags() are both the trim() + split("\s*,\s*", -1) version, so no guard for empty space needed in downstreams
| 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); |
| Map<String, MemberAssignment> members | ||
| ) { | ||
| this.members = Objects.requireNonNull(members); | ||
| this.members = Map.copyOf(members); |
AndrewJSchofield
changed the title
chia7712
left a comment
chia7712
left a comment
There was a problem hiding this comment.
@gabriellefu thanks for this patch.
|
|
||
| /** | ||
| * @return The member assignments keyed by member ID. | ||
| * @return The member assignments keyed by member ID. The map is unmodifiable. |
There was a problem hiding this comment.
It appears the original logic aligns with org.apache.kafka.coordinator.group.api.assignor.GroupAssignment. They are used by different assignor interfaces, so having different "decorations" should be fine.
mjsax
left a comment
mjsax
left a comment
There was a problem hiding this comment.
Thanks for the improvements. Made a pass.
| GroupCoordinatorConfig.STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT); | ||
| return new AssignmentConfigsImpl( | ||
| Integer.parseInt(configs.get(NUM_STANDBY_REPLICAS_CONFIG)), | ||
| Integer.parseInt(numStandbyReplicas), |
There was a problem hiding this comment.
If standby replicas are not set, we first convert the default int into String above, to just parse it back to int again. Should we set this up differently:
String numStandbyReplicasConfig = configs.get(NUM_STANDBY_REPLICAS_CONFIG);
int numStandbyReplicas = numStandbyReplicasConfig != null ? Integer.parseInt(numStandbyReplicasConfig) : GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT;
Or something like this? And pass int numStandbyReplicas here directly.
| long metadataHash, | ||
| int validatedTopologyEpoch, | ||
| Map<String, String> assignmentConfigs, | ||
| Optional<AssignmentConfigsImpl> assignmentConfigs, |
There was a problem hiding this comment.
Why is this an Optional ? Same question elsewhere.
There was a problem hiding this comment.
AssignmentConfigs in StreamsGroupMetadataValue.json defaults to null, so if lastassignmentconfig is never recorded, it's gonna be null
| // them unrecorded. The next heartbeat then compares the effective configs against the defaults, so it | ||
| // only rebalances the group if any effective config differs from its default. | ||
| if (value.lastAssignmentConfigs() == null || value.lastAssignmentConfigs().isEmpty()) { | ||
| streamsGroup.setLastAssignmentConfigs(Optional.empty()); |
There was a problem hiding this comment.
Instead of having Optional type wired thought many classes, could we not just set a "default" object here right away?
Would be good to get input from @squah-confluent or @dajac on this one, to not break anything by accident, and keep the code structure consistent across group types.
There was a problem hiding this comment.
Agreed, I'd like to hear their take on this too before changing it.
3 participants
change: guard the output of assignor to make it unmofiable only
convert assignmentimpl when it's neccessary
Reviewers: Chia-Ping Tsai chia7712@gmail.com, Matthias J. Sax
matthias@confluent.io