Re: [PR] KAFKA-20665: Make the assignment refiner injectable [kafka]

["squah-confluent (via GitHub)" <gi...@apache.org>]

squah-confluent commented on code in PR #23322:
URL: https://github.com/apache/kafka/pull/23322#discussion_r3902880404


##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -535,7 +541,8 @@ public class GroupCoordinatorConfig {
         .define(STREAMS_GROUP_MAX_WARMUP_REPLICAS_CONFIG, INT, STREAMS_GROUP_MAX_WARMUP_REPLICAS_DEFAULT, atLeast(0), MEDIUM, STREAMS_GROUP_MAX_WARMUP_REPLICAS_DOC)
         .define(STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_CONFIG, LIST, STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT, ConfigDef.ValidList.anyNonDuplicateValues(true, false), LOW, STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DOC)
         .define(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG, CLASS, STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_DEFAULT, MEDIUM, STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_DOC)
-        .define(STREAMS_GROUP_ACCEPTABLE_RECOVERY_LAG_CONFIG, LONG, STREAMS_GROUP_ACCEPTABLE_RECOVERY_LAG_DEFAULT, atLeast(0L), MEDIUM, STREAMS_GROUP_ACCEPTABLE_RECOVERY_LAG_DOC);
+        .define(STREAMS_GROUP_ACCEPTABLE_RECOVERY_LAG_CONFIG, LONG, STREAMS_GROUP_ACCEPTABLE_RECOVERY_LAG_DEFAULT, atLeast(0L), MEDIUM, STREAMS_GROUP_ACCEPTABLE_RECOVERY_LAG_DOC)
+        .defineInternal(STREAMS_GROUP_ASSIGNMENT_REFINER_CLASS_CONFIG, CLASS, STREAMS_GROUP_ASSIGNMENT_REFINER_CLASS_DEFAULT, null, LOW, STREAMS_GROUP_ASSIGNMENT_REFINER_CLASS_DOC);

Review Comment:
   Can we try to pick an ordering with respect to acceptable recovery lag?
   
   Above we define the constants before STREAMS_GROUP_ACCEPTABLE_RECOVERY_LAG_CONFIG. Here we define the config after acceptable recovery lag and the getter below is far away from the recovery lag getter.



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -370,6 +372,11 @@ Builder withStreamsGroupAssignors(List<TaskAssignor> streamsGroupAssignors) {
             return this;
         }
 
+        Builder withAssignmentRefiner(AssignmentRefiner assignmentRefiner) {

Review Comment:
   Can we rename this to `withStreamsGroupAssignmentRefiner`?
   
   Same for `GroupMetadataManagerTestContext`.



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -329,6 +330,7 @@ public static class Builder {
         private GroupCoordinatorMetricsShard metrics;
         private Optional<Plugin<Authorizer>> authorizerPlugin = null;
         private List<TaskAssignor> streamsGroupAssignors = null;
+        private AssignmentRefiner assignmentRefiner = null;

Review Comment:
   Can we rename this to `streamsGroupAssignmentRefiner`?
   
   Same for `GroupMetadataManagerTestContext`.



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -1065,6 +1072,21 @@ public boolean isStreamsGroupTopologyDescriptionPluginConfigured() {
         return config.getClass(STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG) != null;
     }
 
+    /**
+     * The assignment refiner to derive the intermediate assignment of a streams group with, or {@code null} if none is
+     * configured, in which case the caller falls back to {@link NoOpAssignmentRefiner}.
+     * <p>
+     * Instantiated on demand rather than held in a field, because a {@code GroupCoordinatorConfig} is constructed
+     * whenever any dynamic broker config changes ({@code DynamicBrokerConfig.processReconfiguration} builds a whole
+     * {@code KafkaConfig}), and we do not want to construct a refiner for each of those. A shard asks for one when it
+     * is loaded, so every shard gets its own instance.
+     */

Review Comment:
   Claude has a habit of oversharing in comments and I am not a fan of it. It's really challenging to trim down since everything it includes is _true_. For method javadocs we can try approaching it from the angle of "what does the caller need to know?" and these items fall out:
   
   * When `null`, it is up to the caller to choose a default. It doesn't make sense to tell the caller that they have chosen `NoOpAssignmentRefiner` as the default.
   * The caller is interested in knowing that each call constructs a fresh instance, since refiners are not thread safe. The part about shards is an implementation detail of the caller.
   * The reasoning about not instantiating in the constructor is an implementation detail and belongs in a comment (but we wouldn't do that anyway since each call wants to return a fresh instance).



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/AssignmentRefiner.java:
##########
@@ -44,11 +45,20 @@
  *     the decisions of a step are taken once, when the epoch is minted, and do not change while the members reconcile
  *     towards it. See {@link StreamsGroup#refinedAssignment(int)}.</li>
  * </ul>
+ * <p>
+ * One instance is created per coordinator shard, and serves every streams group on that shard: the two configurations
+ * a refinement depends on are per-group overridable and are therefore passed in on every call rather than held as
+ * instance state. Since a shard is single-threaded, an implementation does not have to be thread-safe unless it shares
+ * state across shards.
+ * <p>
+ * The default is {@link NoOpAssignmentRefiner}. Which implementation is used is not a public extension point; it is
+ * selected by an internal broker configuration
+ * ({@code GroupCoordinatorConfig.STREAMS_GROUP_ASSIGNMENT_REFINER_CLASS_CONFIG}) that exists so that tests can put a

Review Comment:
   writing style nit: No dramatic reveals please! Let's lead with what is (config decides the refiner) rather than what is not ("not a public extension point").



##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -433,6 +435,10 @@ public class GroupCoordinatorConfig {
     public static final Class<?> STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_DEFAULT = null;
     public static final String STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_DOC = "The fully qualified class name of a StreamsGroupTopologyDescriptionPlugin implementation. When not set, the feature is disabled.";
 
+    public static final String STREAMS_GROUP_ASSIGNMENT_REFINER_CLASS_CONFIG = "group.streams.assignment.refiner.class";
+    public static final Class<?> STREAMS_GROUP_ASSIGNMENT_REFINER_CLASS_DEFAULT = null;

Review Comment:
   I wonder if we should point the config-level default to the real default class (not strongly for or against)



-- 
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