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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
