You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "gabriellefu (via GitHub)" <gi...@apache.org> on 2026/08/05 20:44:30 UTC
[PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
gabriellefu opened a new pull request, #23092:
URL: https://github.com/apache/kafka/pull/23092
Adding AssignmentConfigs a public api for kip1357
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3752280555
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java:
##########
@@ -114,7 +116,22 @@ public TargetAssignmentBuilder(
this.groupId = Objects.requireNonNull(groupId);
this.groupEpoch = groupEpoch;
this.assignor = Objects.requireNonNull(assignor);
- this.assignmentConfigs = Objects.requireNonNull(assignmentConfigs);
+ this.assignmentConfigs = toAssignmentConfigs(Objects.requireNonNull(assignmentConfigs));
+ }
+
+ /**
+ * Converts the raw assignment configs computed for the group into the typed configs passed to the assignor.
+ */
+ private static AssignmentConfigs toAssignmentConfigs(Map<String, String> assignmentConfigs) {
Review Comment:
have changed. 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3759770621
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImpl.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.coordinator.group.api.streams.assignor.AssignmentConfigs;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * The assignment configurations for a streams group.
+ *
+ * @param numStandbyReplicas The number of standby replicas for each task.
+ * @param rackAwareAssignmentTags The client tags used to distribute standby tasks across racks.
+ */
+public record AssignmentConfigsImpl(
+ int numStandbyReplicas,
+ List<String> rackAwareAssignmentTags
+) implements AssignmentConfigs {
+
+ // The names under which the configurations are passed to the assignor and recorded for the group.
+ public static final String NUM_STANDBY_REPLICAS_CONFIG = "num.standby.replicas";
+ public static final String RACK_AWARE_ASSIGNMENT_TAGS_CONFIG = "rack.aware.assignment.tags";
+
+ /**
+ * The configs of a group that has none of them set, holding the default value of every configuration.
+ */
+ public static final AssignmentConfigsImpl DEFAULT = new AssignmentConfigsImpl(
+ GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT,
+ // The parsed form of STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT, which ConfigDef spells as "".
+ List.of()
Review Comment:
I have added a helper here to parse STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3750597472
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignorTest.java:
##########
@@ -926,7 +925,7 @@ public void shouldHandleOddNumberOfClientsWithStandbyTasks() {
}
GroupAssignment result = assignor.assign(
- new GroupSpecImpl(members, mkMap(mkEntry(NUM_STANDBY_REPLICAS_CONFIG, String.valueOf(numStandbyReplicas)))),
+ new GroupSpecImpl(members, new AssignmentConfigsImpl(numStandbyReplicas, List.of())),
Review Comment:
I have another pr is to avoid epoch bump when there's new config being introduced. So I was thinking about implement the default part maybe after this pr is merged? https://github.com/apache/kafka/pull/23088
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3762257819
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImpl.java:
##########
@@ -61,17 +60,24 @@ public static AssignmentConfigsImpl fromMap(Map<String, String> configs) {
if (configs.isEmpty()) {
return DEFAULT;
}
- // The rack-aware assignment tags are only set when any are configured, and are joined from values that
- // ConfigDef has already validated to be non-empty and free of surrounding whitespace.
- String rackAwareAssignmentTags = configs.get(RACK_AWARE_ASSIGNMENT_TAGS_CONFIG);
+ // 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);
return new AssignmentConfigsImpl(
Integer.parseInt(configs.get(NUM_STANDBY_REPLICAS_CONFIG)),
- rackAwareAssignmentTags == null
- ? List.of()
- : List.of(rackAwareAssignmentTags.split(","))
+ parseRackAwareAssignmentTags(rackAwareAssignmentTags)
);
}
+ /**
+ * Parses a recorded rack-aware assignment tags value, the way {@link org.apache.kafka.common.config.ConfigDef}
+ * parses a {@code LIST} configuration: an empty value is an empty list, not a list holding an empty string.
+ */
+ private static List<String> parseRackAwareAssignmentTags(String rackAwareAssignmentTags) {
+ return rackAwareAssignmentTags.isEmpty() ? List.of() : List.of(rackAwareAssignmentTags.split(","));
Review Comment:
the input of this will be the output of String.join(",", tags), in this case, there won't be " ". But I do feel like here the type has been a bit confusing, I will submit a follow up pr to clean up those things
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3751591883
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImpl.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import org.apache.kafka.coordinator.group.api.streams.assignor.AssignmentConfigs;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * The assignment configurations for a streams group.
+ *
+ * @param numStandbyReplicas The number of standby replicas for each task.
+ * @param rackAwareAssignmentTags The client tags used to distribute standby tasks across racks.
+ */
+public record AssignmentConfigsImpl(
+ int numStandbyReplicas,
+ List<String> rackAwareAssignmentTags
+) implements AssignmentConfigs {
+
+ private static final String NUM_STANDBY_REPLICAS_CONFIG = "num.standby.replicas";
+ private static final String RACK_AWARE_ASSIGNMENT_TAGS_CONFIG = "rack.aware.assignment.tags";
+
+ /**
+ * The configs used for a group that has none of them set.
+ */
+ public static final AssignmentConfigsImpl DEFAULT = new AssignmentConfigsImpl(0, List.of());
Review Comment:
i added AssignmentConfigsImpl.DEFAULT which is the default value of all the single config in the configs(), and also added some helper to construct AssignmentConfigsImpl with all default value except certain 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3750973606
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignorTest.java:
##########
@@ -926,7 +925,7 @@ public void shouldHandleOddNumberOfClientsWithStandbyTasks() {
}
GroupAssignment result = assignor.assign(
- new GroupSpecImpl(members, mkMap(mkEntry(NUM_STANDBY_REPLICAS_CONFIG, String.valueOf(numStandbyReplicas)))),
+ new GroupSpecImpl(members, new AssignmentConfigsImpl(numStandbyReplicas, List.of())),
Review Comment:
sounds good, i can do that. In the tests we can only pass the configs we actually care about , otherwise they will use there default value, for replicas is 0 and for tag is "".
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #23092:
URL: https://github.com/apache/kafka/pull/23092
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3749666480
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignorTest.java:
##########
@@ -926,7 +925,7 @@ public void shouldHandleOddNumberOfClientsWithStandbyTasks() {
}
GroupAssignment result = assignor.assign(
- new GroupSpecImpl(members, mkMap(mkEntry(NUM_STANDBY_REPLICAS_CONFIG, String.valueOf(numStandbyReplicas)))),
+ new GroupSpecImpl(members, new AssignmentConfigsImpl(numStandbyReplicas, List.of())),
Review Comment:
These constructor calls are still going to be annoying to update when we add new configs.
We could leave them be or we could try to tidy things up some more, for example, by adding helper methods to construct an `AssignmentConfigsImpl` with default values but a specific `num.standby.replicas` or `rack.aware.assignment.tags` (eg. a `withNumStandbyReplicas`)
(Only applies to calls where we would want to take the defaults for future configs.)
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImpl.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import org.apache.kafka.coordinator.group.api.streams.assignor.AssignmentConfigs;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * The assignment configurations for a streams group.
+ *
+ * @param numStandbyReplicas The number of standby replicas for each task.
+ * @param rackAwareAssignmentTags The client tags used to distribute standby tasks across racks.
+ */
+public record AssignmentConfigsImpl(
+ int numStandbyReplicas,
+ List<String> rackAwareAssignmentTags
+) implements AssignmentConfigs {
+
+ private static final String NUM_STANDBY_REPLICAS_CONFIG = "num.standby.replicas";
+ private static final String RACK_AWARE_ASSIGNMENT_TAGS_CONFIG = "rack.aware.assignment.tags";
+
+ /**
+ * The configs used for a group that has none of them set.
+ */
+ public static final AssignmentConfigsImpl DEFAULT = new AssignmentConfigsImpl(0, List.of());
Review Comment:
Where do we plan to define the default `num.standby.replicas` and `rack.aware.assignment.tags`? Can we have a single source of truth for the defaults?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3754416057
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/MockTaskAssignor.java:
##########
@@ -31,7 +32,7 @@ public class MockTaskAssignor implements TaskAssignor {
private final String name;
private GroupAssignment preparedGroupAssignment = null;
- private Map<String, String> assignmentConfigs = Map.of();
+ private AssignmentConfigs assignmentConfigs = null;
Review Comment:
Why are we setting `null`? Would it be better to use `DEFAULT` ?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #23092:
URL: https://github.com/apache/kafka/pull/23092#issuecomment-5271758401
Merged to `trunk` and cherry-picked to `4.4` branch.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3773441242
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImpl.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.coordinator.group.api.streams.assignor.AssignmentConfigs;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * The assignment configurations for a streams group.
+ *
+ * @param numStandbyReplicas The number of standby replicas for each task.
+ * @param rackAwareAssignmentTags The client tags used to distribute standby tasks across racks.
+ */
+public record AssignmentConfigsImpl(
+ int numStandbyReplicas,
+ List<String> rackAwareAssignmentTags
+) implements AssignmentConfigs {
+
+ // The names under which the configurations are passed to the assignor and recorded for the group.
+ public static final String NUM_STANDBY_REPLICAS_CONFIG = "num.standby.replicas";
+ public static final String RACK_AWARE_ASSIGNMENT_TAGS_CONFIG = "rack.aware.assignment.tags";
+
+ /**
+ * The configs of a group that has none of them set, holding the default value of every configuration.
+ */
+ public static final AssignmentConfigsImpl DEFAULT = new AssignmentConfigsImpl(
+ GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT,
+ // The parsed form of STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT, which ConfigDef spells as "".
+ List.of()
+ );
+
+ public AssignmentConfigsImpl {
+ // The list is exposed to a custom assignor through the public AssignmentConfigs interface.
+ rackAwareAssignmentTags = List.copyOf(Objects.requireNonNull(rackAwareAssignmentTags));
+ }
+
+ /**
+ * Converts the raw assignment configs computed 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 set when any are configured, and are joined from values that
+ // ConfigDef has already validated to be non-empty and free of surrounding whitespace.
+ String rackAwareAssignmentTags = configs.get(RACK_AWARE_ASSIGNMENT_TAGS_CONFIG);
+ return new AssignmentConfigsImpl(
+ Integer.parseInt(configs.get(NUM_STANDBY_REPLICAS_CONFIG)),
Review Comment:
I have a similar question here. Maybe we could leverage the default value?
```java
String numStandby = configs.getOrDefault(
NUM_STANDBY_REPLICAS_CONFIG,
String.valueOf(GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT)
);
return new AssignmentConfigsImpl(
Integer.parseInt(numStandby),
parseRackAwareAssignmentTags(rackAwareAssignmentTags)
);
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3754375195
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImpl.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.coordinator.group.api.streams.assignor.AssignmentConfigs;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * The assignment configurations for a streams group.
+ *
+ * @param numStandbyReplicas The number of standby replicas for each task.
+ * @param rackAwareAssignmentTags The client tags used to distribute standby tasks across racks.
+ */
+public record AssignmentConfigsImpl(
+ int numStandbyReplicas,
+ List<String> rackAwareAssignmentTags
+) implements AssignmentConfigs {
+
+ // The names under which the configurations are passed to the assignor and recorded for the group.
+ public static final String NUM_STANDBY_REPLICAS_CONFIG = "num.standby.replicas";
+ public static final String RACK_AWARE_ASSIGNMENT_TAGS_CONFIG = "rack.aware.assignment.tags";
+
+ /**
+ * The configs of a group that has none of them set, holding the default value of every configuration.
+ */
+ public static final AssignmentConfigsImpl DEFAULT = new AssignmentConfigsImpl(
+ GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT,
+ // The parsed form of STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT, which ConfigDef spells as "".
+ List.of()
+ );
+
+ public AssignmentConfigsImpl {
+ // The list is exposed to a custom assignor through the public AssignmentConfigs interface.
+ rackAwareAssignmentTags = List.copyOf(Objects.requireNonNull(rackAwareAssignmentTags));
+ }
+
+ /**
+ * Converts the raw assignment configs computed 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 set when any are configured, and are joined from values that
+ // ConfigDef has already validated to be non-empty and free of surrounding whitespace.
+ String rackAwareAssignmentTags = configs.get(RACK_AWARE_ASSIGNMENT_TAGS_CONFIG);
+ return new AssignmentConfigsImpl(
+ Integer.parseInt(configs.get(NUM_STANDBY_REPLICAS_CONFIG)),
Review Comment:
Don't we need `null` check here and translate it to `0` ?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3761097508
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImpl.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.coordinator.group.api.streams.assignor.AssignmentConfigs;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * The assignment configurations for a streams group.
+ *
+ * @param numStandbyReplicas The number of standby replicas for each task.
+ * @param rackAwareAssignmentTags The client tags used to distribute standby tasks across racks.
+ */
+public record AssignmentConfigsImpl(
+ int numStandbyReplicas,
+ List<String> rackAwareAssignmentTags
+) implements AssignmentConfigs {
+
+ // The names under which the configurations are passed to the assignor and recorded for the group.
+ public static final String NUM_STANDBY_REPLICAS_CONFIG = "num.standby.replicas";
+ public static final String RACK_AWARE_ASSIGNMENT_TAGS_CONFIG = "rack.aware.assignment.tags";
+
+ /**
+ * The configs of a group that has none of them set, holding the default value of every configuration.
+ */
+ public static final AssignmentConfigsImpl DEFAULT = new AssignmentConfigsImpl(
+ GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT,
+ // The parsed form of STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT, which ConfigDef spells as "".
+ List.of()
+ );
+
+ public AssignmentConfigsImpl {
+ // The list is exposed to a custom assignor through the public AssignmentConfigs interface.
+ rackAwareAssignmentTags = List.copyOf(Objects.requireNonNull(rackAwareAssignmentTags));
+ }
+
+ /**
+ * Converts the raw assignment configs computed 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 set when any are configured, and are joined from values that
+ // ConfigDef has already validated to be non-empty and free of surrounding whitespace.
+ String rackAwareAssignmentTags = configs.get(RACK_AWARE_ASSIGNMENT_TAGS_CONFIG);
+ return new AssignmentConfigsImpl(
+ Integer.parseInt(configs.get(NUM_STANDBY_REPLICAS_CONFIG)),
Review Comment:
Ok. Seems we are scattered logic, making it hard to reason about this... No critical for 4.4 release, but wondering if we could do a follow up cleanup PR, streamlining this a little bit better to make it easier as a (human 🤣) reviewer to follow...
Maybe we should "unify" the raw `Map<String, String>` and `AssignmentConfigsImpl` somehow, to have a single object we use and pass around, an avoid converting the one into the other multiple times along the way, at different places.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3759691739
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/MockTaskAssignor.java:
##########
@@ -31,7 +32,7 @@ public class MockTaskAssignor implements TaskAssignor {
private final String name;
private GroupAssignment preparedGroupAssignment = null;
- private Map<String, String> assignmentConfigs = Map.of();
+ private AssignmentConfigs assignmentConfigs = null;
Review Comment:
changed
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3750624824
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignorTest.java:
##########
@@ -926,7 +925,7 @@ public void shouldHandleOddNumberOfClientsWithStandbyTasks() {
}
GroupAssignment result = assignor.assign(
- new GroupSpecImpl(members, mkMap(mkEntry(NUM_STANDBY_REPLICAS_CONFIG, String.valueOf(numStandbyReplicas)))),
+ new GroupSpecImpl(members, new AssignmentConfigsImpl(numStandbyReplicas, List.of())),
Review Comment:
I can close the other pr and put the change in this pr too
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3755576986
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImpl.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.coordinator.group.api.streams.assignor.AssignmentConfigs;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * The assignment configurations for a streams group.
+ *
+ * @param numStandbyReplicas The number of standby replicas for each task.
+ * @param rackAwareAssignmentTags The client tags used to distribute standby tasks across racks.
+ */
+public record AssignmentConfigsImpl(
+ int numStandbyReplicas,
+ List<String> rackAwareAssignmentTags
+) implements AssignmentConfigs {
+
+ // The names under which the configurations are passed to the assignor and recorded for the group.
+ public static final String NUM_STANDBY_REPLICAS_CONFIG = "num.standby.replicas";
+ public static final String RACK_AWARE_ASSIGNMENT_TAGS_CONFIG = "rack.aware.assignment.tags";
+
+ /**
+ * The configs of a group that has none of them set, holding the default value of every configuration.
+ */
+ public static final AssignmentConfigsImpl DEFAULT = new AssignmentConfigsImpl(
+ GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT,
+ // The parsed form of STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT, which ConfigDef spells as "".
+ List.of()
Review Comment:
Hmmm... This still seems to be error prone? In the end, we should have a single source of truth... Should we rather use `STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT` and parse it into empty List? If the default changes, this code would update immediatly?
But on the other hand, it seem we are using `DEFAULT` only in tests, what make it somewhat questionable, if it's the right place to add it here?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3750741940
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/StickyTaskAssignorTest.java:
##########
@@ -926,7 +925,7 @@ public void shouldHandleOddNumberOfClientsWithStandbyTasks() {
}
GroupAssignment result = assignor.assign(
- new GroupSpecImpl(members, mkMap(mkEntry(NUM_STANDBY_REPLICAS_CONFIG, String.valueOf(numStandbyReplicas)))),
+ new GroupSpecImpl(members, new AssignmentConfigsImpl(numStandbyReplicas, List.of())),
Review Comment:
hm I think the not-bumping logic belongs in separate PR. Maybe we could bring the constant definition forward? But I don't really mind as long as the final state is clean.
How do you imagine these `new AssignmentConfigsImpl` calls to look at the end of it all?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3752242820
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImpl.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.coordinator.group.api.streams.assignor.AssignmentConfigs;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * The assignment configurations for a streams group.
+ *
+ * @param numStandbyReplicas The number of standby replicas for each task.
+ * @param rackAwareAssignmentTags The client tags used to distribute standby tasks across racks.
+ */
+public record AssignmentConfigsImpl(
+ int numStandbyReplicas,
+ List<String> rackAwareAssignmentTags
+) implements AssignmentConfigs {
+
+ // The names under which the configurations are passed to the assignor and recorded for the group.
+ public static final String NUM_STANDBY_REPLICAS_CONFIG = "num.standby.replicas";
+ public static final String RACK_AWARE_ASSIGNMENT_TAGS_CONFIG = "rack.aware.assignment.tags";
+
+ /**
+ * The configs of a group that has none of them set, holding the default value of every configuration.
+ */
+ public static final AssignmentConfigsImpl DEFAULT = new AssignmentConfigsImpl(
+ GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT,
+ // The parsed form of STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT, which ConfigDef spells as "".
+ List.of()
Review Comment:
the default of "rack.aware.assignment.tags" is "", and if we directly parse it into a List, it will give a [""] instead of a empty list [], so we still hardcode the default here
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3768069080
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImplTest.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class AssignmentConfigsImplTest {
+
+ @Test
+ void testDefaultHoldsTheDefaultOfEveryConfig() {
+ assertEquals(GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT,
+ AssignmentConfigsImpl.DEFAULT.numStandbyReplicas());
+ assertEquals(List.of(), AssignmentConfigsImpl.DEFAULT.rackAwareAssignmentTags());
+ }
+
+ @Test
+ void testFromEmptyMap() {
+ // A group metadata record written before the last assignment configs were persisted replays as an empty map.
+ assertEquals(AssignmentConfigsImpl.DEFAULT, AssignmentConfigsImpl.fromMap(Map.of()));
+ }
+
+ @Test
+ void testFromMapWithoutRackAwareAssignmentTags() {
+ // The tags are only put in the map when any are configured.
+ assertEquals(
+ AssignmentConfigsImpl.DEFAULT.withNumStandbyReplicas(2),
+ AssignmentConfigsImpl.fromMap(Map.of("num.standby.replicas", "2"))
+ );
+ }
+
+ @Test
+ void testFromMap() {
+ assertEquals(
+ AssignmentConfigsImpl.DEFAULT
+ .withNumStandbyReplicas(1)
+ .withRackAwareAssignmentTags(List.of("tag1", "tag2")),
+ AssignmentConfigsImpl.fromMap(Map.of(
+ "num.standby.replicas", "1",
+ "rack.aware.assignment.tags", "tag1,tag2"
+ ))
+ );
+ }
+
+ @Test
+ void testWithersLeaveTheOtherConfigsUntouched() {
Review Comment:
have changed! 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3787048322
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImpl.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.coordinator.group.api.streams.assignor.AssignmentConfigs;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * The assignment configurations for a streams group.
+ *
+ * @param numStandbyReplicas The number of standby replicas for each task.
+ * @param rackAwareAssignmentTags The client tags used to distribute standby tasks across racks.
+ */
+public record AssignmentConfigsImpl(
+ int numStandbyReplicas,
+ List<String> rackAwareAssignmentTags
+) implements AssignmentConfigs {
+
+ // The names under which the configurations are passed to the assignor and recorded for the group.
+ public static final String NUM_STANDBY_REPLICAS_CONFIG = "num.standby.replicas";
+ public static final String RACK_AWARE_ASSIGNMENT_TAGS_CONFIG = "rack.aware.assignment.tags";
+
+ /**
+ * The configs of a group that has none of them set, holding the default value of every configuration.
+ */
+ public static final AssignmentConfigsImpl DEFAULT = new AssignmentConfigsImpl(
+ GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT,
+ // The parsed form of STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT, which ConfigDef spells as "".
+ List.of()
+ );
+
+ public AssignmentConfigsImpl {
+ // The list is exposed to a custom assignor through the public AssignmentConfigs interface.
+ rackAwareAssignmentTags = List.copyOf(Objects.requireNonNull(rackAwareAssignmentTags));
+ }
+
+ /**
+ * Converts the raw assignment configs computed 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 set when any are configured, and are joined from values that
+ // ConfigDef has already validated to be non-empty and free of surrounding whitespace.
+ String rackAwareAssignmentTags = configs.get(RACK_AWARE_ASSIGNMENT_TAGS_CONFIG);
+ return new AssignmentConfigsImpl(
+ Integer.parseInt(configs.get(NUM_STANDBY_REPLICAS_CONFIG)),
Review Comment:
have updated in this following pr:https://github.com/apache/kafka/pull/23165 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3765413292
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImplTest.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class AssignmentConfigsImplTest {
+
+ @Test
+ void testDefaultHoldsTheDefaultOfEveryConfig() {
+ assertEquals(GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT,
+ AssignmentConfigsImpl.DEFAULT.numStandbyReplicas());
+ assertEquals(List.of(), AssignmentConfigsImpl.DEFAULT.rackAwareAssignmentTags());
+ }
+
+ @Test
+ void testFromEmptyMap() {
+ // A group metadata record written before the last assignment configs were persisted replays as an empty map.
+ assertEquals(AssignmentConfigsImpl.DEFAULT, AssignmentConfigsImpl.fromMap(Map.of()));
+ }
+
+ @Test
+ void testFromMapWithoutRackAwareAssignmentTags() {
+ // The tags are only put in the map when any are configured.
+ assertEquals(
+ AssignmentConfigsImpl.DEFAULT.withNumStandbyReplicas(2),
+ AssignmentConfigsImpl.fromMap(Map.of("num.standby.replicas", "2"))
+ );
+ }
+
+ @Test
+ void testFromMap() {
+ assertEquals(
+ AssignmentConfigsImpl.DEFAULT
+ .withNumStandbyReplicas(1)
+ .withRackAwareAssignmentTags(List.of("tag1", "tag2")),
+ AssignmentConfigsImpl.fromMap(Map.of(
+ "num.standby.replicas", "1",
+ "rack.aware.assignment.tags", "tag1,tag2"
+ ))
+ );
+ }
+
+ @Test
+ void testWithersLeaveTheOtherConfigsUntouched() {
Review Comment:
nit: I'd just call this `testWithers`
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImplTest.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class AssignmentConfigsImplTest {
+
+ @Test
+ void testDefaultHoldsTheDefaultOfEveryConfig() {
Review Comment:
nit: I'd just call this `testDefault`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3759004827
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImpl.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.coordinator.group.api.streams.assignor.AssignmentConfigs;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * The assignment configurations for a streams group.
+ *
+ * @param numStandbyReplicas The number of standby replicas for each task.
+ * @param rackAwareAssignmentTags The client tags used to distribute standby tasks across racks.
+ */
+public record AssignmentConfigsImpl(
+ int numStandbyReplicas,
+ List<String> rackAwareAssignmentTags
+) implements AssignmentConfigs {
+
+ // The names under which the configurations are passed to the assignor and recorded for the group.
+ public static final String NUM_STANDBY_REPLICAS_CONFIG = "num.standby.replicas";
+ public static final String RACK_AWARE_ASSIGNMENT_TAGS_CONFIG = "rack.aware.assignment.tags";
+
+ /**
+ * The configs of a group that has none of them set, holding the default value of every configuration.
+ */
+ public static final AssignmentConfigsImpl DEFAULT = new AssignmentConfigsImpl(
+ GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT,
+ // The parsed form of STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT, which ConfigDef spells as "".
+ List.of()
Review Comment:
we also use DEFAULT in fromMap(), if the config is a empty map, it returns DEFAULT otherwise it parse the config. But I agree with the STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT part I will update accordingly.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "gabriellefu (via GitHub)" <gi...@apache.org>.
gabriellefu commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3759662258
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImpl.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams.assignor;
+
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.coordinator.group.api.streams.assignor.AssignmentConfigs;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * The assignment configurations for a streams group.
+ *
+ * @param numStandbyReplicas The number of standby replicas for each task.
+ * @param rackAwareAssignmentTags The client tags used to distribute standby tasks across racks.
+ */
+public record AssignmentConfigsImpl(
+ int numStandbyReplicas,
+ List<String> rackAwareAssignmentTags
+) implements AssignmentConfigs {
+
+ // The names under which the configurations are passed to the assignor and recorded for the group.
+ public static final String NUM_STANDBY_REPLICAS_CONFIG = "num.standby.replicas";
+ public static final String RACK_AWARE_ASSIGNMENT_TAGS_CONFIG = "rack.aware.assignment.tags";
+
+ /**
+ * The configs of a group that has none of them set, holding the default value of every configuration.
+ */
+ public static final AssignmentConfigsImpl DEFAULT = new AssignmentConfigsImpl(
+ GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_DEFAULT,
+ // The parsed form of STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT, which ConfigDef spells as "".
+ List.of()
+ );
+
+ public AssignmentConfigsImpl {
+ // The list is exposed to a custom assignor through the public AssignmentConfigs interface.
+ rackAwareAssignmentTags = List.copyOf(Objects.requireNonNull(rackAwareAssignmentTags));
+ }
+
+ /**
+ * Converts the raw assignment configs computed 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 set when any are configured, and are joined from values that
+ // ConfigDef has already validated to be non-empty and free of surrounding whitespace.
+ String rackAwareAssignmentTags = configs.get(RACK_AWARE_ASSIGNMENT_TAGS_CONFIG);
+ return new AssignmentConfigsImpl(
+ Integer.parseInt(configs.get(NUM_STANDBY_REPLICAS_CONFIG)),
Review Comment:
In the original code, for replicas, there are two path, one is the whole map is empty, then it's ok to not having replicas set and here we return DEFAULT, other than that, the replicas should always be in the config
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/MockTaskAssignor.java:
##########
@@ -31,7 +32,7 @@ public class MockTaskAssignor implements TaskAssignor {
private final String name;
private GroupAssignment preparedGroupAssignment = null;
- private Map<String, String> assignmentConfigs = Map.of();
+ private AssignmentConfigs assignmentConfigs = null;
Review Comment:
oh my mistake, cause it will be equalavent to DEFAULT before as a empty map, I will update here
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3761081117
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/assignor/AssignmentConfigsImpl.java:
##########
@@ -61,17 +60,24 @@ public static AssignmentConfigsImpl fromMap(Map<String, String> configs) {
if (configs.isEmpty()) {
return DEFAULT;
}
- // The rack-aware assignment tags are only set when any are configured, and are joined from values that
- // ConfigDef has already validated to be non-empty and free of surrounding whitespace.
- String rackAwareAssignmentTags = configs.get(RACK_AWARE_ASSIGNMENT_TAGS_CONFIG);
+ // 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);
return new AssignmentConfigsImpl(
Integer.parseInt(configs.get(NUM_STANDBY_REPLICAS_CONFIG)),
- rackAwareAssignmentTags == null
- ? List.of()
- : List.of(rackAwareAssignmentTags.split(","))
+ parseRackAwareAssignmentTags(rackAwareAssignmentTags)
);
}
+ /**
+ * Parses a recorded rack-aware assignment tags value, the way {@link org.apache.kafka.common.config.ConfigDef}
+ * parses a {@code LIST} configuration: an empty value is an empty list, not a list holding an empty string.
+ */
+ private static List<String> parseRackAwareAssignmentTags(String rackAwareAssignmentTags) {
+ return rackAwareAssignmentTags.isEmpty() ? List.of() : List.of(rackAwareAssignmentTags.split(","));
Review Comment:
`isEmpty()` checks for length() == 0, right? What about `" "` (or similar). Do we need to to do rackAwareAssignmentTags.trim().isEmpty()` ?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20790: add AssignmentConfigs interface [kafka]
Posted by "squah-confluent (via GitHub)" <gi...@apache.org>.
squah-confluent commented on code in PR #23092:
URL: https://github.com/apache/kafka/pull/23092#discussion_r3729027720
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TargetAssignmentBuilder.java:
##########
@@ -114,7 +116,22 @@ public TargetAssignmentBuilder(
this.groupId = Objects.requireNonNull(groupId);
this.groupEpoch = groupEpoch;
this.assignor = Objects.requireNonNull(assignor);
- this.assignmentConfigs = Objects.requireNonNull(assignmentConfigs);
+ this.assignmentConfigs = toAssignmentConfigs(Objects.requireNonNull(assignmentConfigs));
+ }
+
+ /**
+ * Converts the raw assignment configs computed for the group into the typed configs passed to the assignor.
+ */
+ private static AssignmentConfigs toAssignmentConfigs(Map<String, String> assignmentConfigs) {
Review Comment:
A better place for this method could be as a constructor or static method on `AssignmentConfigsImpl`.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/assignor/MockAssignorTest.java:
##########
@@ -54,7 +53,7 @@ public void testZeroMembers() {
TaskAssignorException ex = assertThrows(TaskAssignorException.class, () -> assignor.assign(
new GroupSpecImpl(
Map.of(),
- new HashMap<>()
+ new AssignmentConfigsImpl(0, List.of())
Review Comment:
It's going to be really annoying to update all these tests when we add new assignment configs.
We could consider declaring `AssignmentConfigsImpl.DEFAULT` or adding a constructor `AssignmentConfigsImpl(Map<String, String>)`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org