chia7712 commented on code in PR #23165:
URL: https://github.com/apache/kafka/pull/23165#discussion_r3794532019
##########
group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/assignor/GroupAssignment.java:
##########
@@ -36,11 +35,11 @@ public class GroupAssignment {
public GroupAssignment(
Map<String, MemberAssignment> members
) {
- this.members = Objects.requireNonNull(members);
+ this.members = Map.copyOf(members);
}
/**
- * @return The member assignments keyed by member ID.
+ * @return The member assignments keyed by member ID. The map is
unmodifiable.
Review 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.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsCoordinatorRecordHelpers.java:
##########
@@ -101,18 +103,21 @@ public static CoordinatorRecord
newStreamsGroupMetadataRecord(
int newGroupEpoch,
long metadataHash,
int validatedTopologyEpoch,
- Map<String, String> assignmentConfigs,
+ Optional<AssignmentConfigsImpl> assignmentConfigs,
int storedDescriptionTopologyEpoch,
int failedDescriptionTopologyEpoch
) {
Objects.requireNonNull(groupId, "groupId should not be null here");
Objects.requireNonNull(assignmentConfigs, "assignmentConfigs should
not be null here");
- List<StreamsGroupMetadataValue.LastAssignmentConfig>
assignmentConfigList = assignmentConfigs.entrySet().stream()
- .map(entry -> new StreamsGroupMetadataValue.LastAssignmentConfig()
- .setKey(entry.getKey())
- .setValue(entry.getValue()))
- .toList();
+ // Configs that were never recorded stay unrecorded (an empty list);
the epoch bump check treats an
+ // unrecorded group as running the defaults.
+ List<StreamsGroupMetadataValue.LastAssignmentConfig>
assignmentConfigList =
+
assignmentConfigs.map(AssignmentConfigsImpl::toMap).orElse(Map.of()).entrySet().stream()
Review Comment:
It seems `AssignmentConfigsImpl::toMap` focus us to use the implementation
`AssignmentConfigsImpl` rather than the interface `AssignmentConfigs` in the
codebase. Maybe `AssignmentConfigsImpl#toMap` could be a helper method and then
we could pass `AssignmentConfigs` instead?
```java
public static Map<String, String> toMap(AssignmentConfigs
assignmentConfigs) {
// The rack-aware assignment tags are only recorded when any are
configured, matching what fromMap expects.
Map<String, String> configs = new TreeMap<>();
configs.put(NUM_STANDBY_REPLICAS_CONFIG,
Integer.toString(assignmentConfigs.numStandbyReplicas()));
if (!assignmentConfigs.rackAwareAssignmentTags().isEmpty()) {
configs.put(RACK_AWARE_ASSIGNMENT_TAGS_CONFIG, String.join(",",
assignmentConfigs.rackAwareAssignmentTags()));
}
return configs;
}
```
--
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]
