mjsax commented on code in PR #22213:
URL: https://github.com/apache/kafka/pull/22213#discussion_r3574798501
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java:
##########
@@ -297,6 +303,12 @@ public final class GroupConfig extends AbstractConfig {
atLeast(0),
MEDIUM,
GroupCoordinatorConfig.STREAMS_GROUP_NUM_WARMUP_REPLICAS_DOC)
+ .define(STREAMS_RACK_AWARE_ASSIGNMENT_TAGS_CONFIG,
+ LIST,
+
GroupCoordinatorConfig.STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT,
+ ConfigDef.ValidList.anyNonDuplicateValues(true, false),
+ MEDIUM,
Review Comment:
I think this should be LOW, as it is LOW in `StreamsConfig`, too.
We are just cleaning up these importance levels:
https://github.com/apache/kafka/pull/22541 \cc @suzhiking -- I think we did
agree on LOW for this one, right?
##########
docs/getting-started/upgrade.md:
##########
@@ -99,6 +99,7 @@ Note: Apache Kafka 4.2 only supports KRaft mode - ZooKeeper
mode has been remove
### Notable changes in 4.2.0
+
Review Comment:
?
Not sure if we need to update this file? This feature is not large enough
for the top-level upgrade guide IMHO.
However, we could add something to `docs/streams/upgrade-guide.md` to report
progress on KIP-1071?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2354,6 +2357,27 @@ private CoordinatorResult<StreamsGroupHeartbeatResult,
CoordinatorRecord> stream
)
));
+ String rackAwareTagsValue =
currentAssignmentConfigs.getOrDefault("rack.aware.assignment.tags", "");
+ // The MISSING_CLIENT_TAGS status (code 6) requires version 1 of the
RPC: version 0 clients
+ // throw on unknown status codes, so it must not be sent to them.
+ if (requestApiVersion >= 1 && !rackAwareTagsValue.isEmpty()) {
+ List<String> requiredTags =
Arrays.asList(rackAwareTagsValue.split(",", -1));
Review Comment:
```suggestion
List<String> requiredTags =
Arrays.asList(rackAwareTagsValue.split("\\s*,\\s*", -1));
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -2354,6 +2357,27 @@ private CoordinatorResult<StreamsGroupHeartbeatResult,
CoordinatorRecord> stream
)
));
+ String rackAwareTagsValue =
currentAssignmentConfigs.getOrDefault("rack.aware.assignment.tags", "");
+ // The MISSING_CLIENT_TAGS status (code 6) requires version 1 of the
RPC: version 0 clients
+ // throw on unknown status codes, so it must not be sent to them.
+ if (requestApiVersion >= 1 && !rackAwareTagsValue.isEmpty()) {
+ List<String> requiredTags =
Arrays.asList(rackAwareTagsValue.split(",", -1));
Review Comment:
I think we also need to strip whitespaces here?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -386,6 +386,10 @@ public class GroupCoordinatorConfig {
public static final String STREAMS_GROUP_MAX_ASSIGNMENT_INTERVAL_MS_DOC =
"The maximum interval between assignment updates for a streams group.";
public static final int STREAMS_GROUP_MAX_ASSIGNMENT_INTERVAL_MS_DEFAULT =
15000;
+ public static final String STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_CONFIG
= "group.streams.rack.aware.assignment.tags";
+ public static final String
STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT = "";
+ public static final String STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DOC =
"List of client tag keys used to distribute standby replicas across Kafka
Streams instances. When configured, the broker-side assignor will make a
best-effort to distribute standby tasks over each client tag dimension.";
Review Comment:
```suggestion
public static final String STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DOC
= "List of client tag keys used to distribute standby replicas across Kafka
Streams instances. When configured, and the used broker-side assignor supports
it, it will make a best-effort to distribute standby tasks over each client tag
dimension.";
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -754,6 +766,27 @@ private void checkConstraints() {
String.format("%s must be greater than or equal to %s",
STREAMS_GROUP_TASK_OFFSET_INTERVAL_MS_CONFIG,
STREAMS_GROUP_MIN_TASK_OFFSET_INTERVAL_MS_CONFIG));
require(streamsGroupNumWarmupReplicas <= streamsGroupMaxWarmupReplicas,
String.format("%s must be less than or equal to %s",
STREAMS_GROUP_NUM_WARMUP_REPLICAS_CONFIG,
STREAMS_GROUP_MAX_WARMUP_REPLICAS_CONFIG));
+
+ // ConfigDef silently de-duplicates LIST values during parsing, so
inspect the raw value to make
+ // sure a broker configured with duplicate rack-aware assignment tags
refuses to start.
+ List<String> rawRackAwareAssignmentTags = rawRackAwareAssignmentTags();
+ require(Set.copyOf(rawRackAwareAssignmentTags).size() ==
rawRackAwareAssignmentTags.size(),
+ String.format("%s must not contain duplicate tag keys.",
STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_CONFIG));
+ }
+
+ /**
+ * Returns the rack-aware assignment tags exactly as configured, before
{@link ConfigDef} removes duplicates.
+ */
+ private List<String> rawRackAwareAssignmentTags() {
+ Object rawValue =
config.originals().get(STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_CONFIG);
Review Comment:
Similar question -- can't we just case this to `String` -- for which case it
won't be `String` type?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupConfig.java:
##########
@@ -467,6 +488,30 @@ public static void validate(
);
}
+ /**
+ * Rejects an alterConfigs request whose rack-aware assignment tags
contain duplicate keys, inspecting the
+ * raw value because {@link ConfigDef} removes duplicates from LIST values
before they can be validated.
+ *
+ * @param newGroupConfig The new unparsed group config overrides.
+ */
+ private static void validateNoDuplicateRackAwareAssignmentTags(Map<String,
?> newGroupConfig) {
Review Comment:
Tracing the code backward, it seems this should be type `Map<String,
String>` and thus we can simplify the logic below?
To make it work, we need to also change the type of `validate(...)` above:
```
public static void validate(
Map<String, String> newGroupConfig,
...
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java:
##########
@@ -603,6 +603,31 @@ private void testStreamsOtherConfigs() {
configs.put(GroupCoordinatorConfig.STREAMS_GROUP_MAX_WARMUP_REPLICAS_CONFIG,
-1);
assertEquals("Invalid value -1 for configuration
group.streams.max.warmup.replicas: Value must be at least 0",
assertThrows(ConfigException.class, () ->
createConfig(configs)).getMessage());
+
+
+ // group.streams.rack.aware.assignment.tags
+
+ // default is empty list
+ configs.clear();
+ GroupCoordinatorConfig defaultTagsConfig = createConfig(configs);
+ assertEquals(List.of(),
defaultTagsConfig.streamsGroupRackAwareAssignmentTags());
+
+ // can parse a non-empty value
+ configs.clear();
+
configs.put(GroupCoordinatorConfig.STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_CONFIG,
"zone,cluster");
+ GroupCoordinatorConfig nonEmptyTagsConfig = createConfig(configs);
+ assertEquals(List.of("zone", "cluster"),
nonEmptyTagsConfig.streamsGroupRackAwareAssignmentTags());
+
+ // rejects empty entries in the list
+ configs.clear();
+
configs.put(GroupCoordinatorConfig.STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_CONFIG,
"zone, ");
+ assertThrows(ConfigException.class, () -> createConfig(configs));
Review Comment:
Should we verify the error message?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupConfigTest.java:
##########
@@ -345,6 +350,19 @@ public void testInvalidProps() {
doTestInvalidProps(props, ConfigException.class);
}
+ @Test
+ public void testStreamsRackAwareAssignmentTagsValidation() {
+ // Duplicate rack-aware assignment tags are rejected rather than
silently de-duplicated.
+ Map<String, String> props = createValidGroupConfig();
+ props.put(GroupConfig.STREAMS_RACK_AWARE_ASSIGNMENT_TAGS_CONFIG,
"zone,zone");
+ doTestInvalidProps(props, InvalidConfigurationException.class);
+
+ // Distinct rack-aware assignment tags are accepted.
+ props = createValidGroupConfig();
+ props.put(GroupConfig.STREAMS_RACK_AWARE_ASSIGNMENT_TAGS_CONFIG,
"zone,cluster");
+ doTestValidProps(props);
+ }
Review Comment:
It might be good to add a few more case for whitespaces? For example `" zone
, cluster "` should parse and return two tags `zone` and `cluster` w/o any
surrounding whitespaces, right?
Also for the default coordinator config test, not just the group config test
here.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -29405,9 +29669,8 @@ private Map<Uuid,
DeleteShareGroupOffsetsResponseData.DeleteShareGroupOffsetsRes
* This matches what streamsGroupAssignmentConfigs() would return.
*/
private Map<String, String> getDefaultAssignmentConfigs() {
- // Use the same default value as
GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT
- return new TreeMap<>(Map.of(
- "num.standby.replicas",
String.valueOf(GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT)
- ));
+ Map<String, String> configs = new TreeMap<>();
+ configs.put("num.standby.replicas",
String.valueOf(GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT));
+ return configs;
Review Comment:
Seems this change is not necessary?
--
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]
