mjsax commented on code in PR #23165:
URL: https://github.com/apache/kafka/pull/23165#discussion_r3818067945
##########
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,
Review Comment:
Why is this an `Optional` ? Same question elsewhere.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -6235,16 +6233,19 @@ public void replay(
streamsGroup.setStoredDescriptionTopologyEpoch(value.storedDescriptionTopologyEpoch());
streamsGroup.setFailedDescriptionTopologyEpoch(value.failedDescriptionTopologyEpoch());
- if (value.lastAssignmentConfigs() != null) {
- streamsGroup.setLastAssignmentConfigs(
+ // A record without configs (written before they were persisted,
or re-persisting such a state) leaves
+ // 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());
Review 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.
##########
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:
I am fine either way. No preference.
##########
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:
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.
--
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]
