KAFKA-20790: add AssignmentConfigs interface - #23092
Conversation
github-actions
Bot
added
triage
squah-confluent
left a comment
squah-confluent
left a comment
There was a problem hiding this comment.
Thanks for updating the PR!
| /** | ||
| * The configs used for a group that has none of them set. | ||
| */ | ||
| public static final AssignmentConfigsImpl DEFAULT = new AssignmentConfigsImpl(0, List.of()); |
There was a problem hiding this 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?
There was a problem hiding this 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
|
|
||
| GroupAssignment result = assignor.assign( | ||
| new GroupSpecImpl(members, mkMap(mkEntry(NUM_STANDBY_REPLICAS_CONFIG, String.valueOf(numStandbyReplicas)))), | ||
| new GroupSpecImpl(members, new AssignmentConfigsImpl(numStandbyReplicas, List.of())), |
There was a problem hiding this 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.)
There was a problem hiding this 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? #23088
There was a problem hiding this comment.
I can close the other pr and put the change in this pr too
There was a problem hiding this 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?
There was a problem hiding this 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 "".
| 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() |
There was a problem hiding this 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
There was a problem hiding this 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?
There was a problem hiding this 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.
There was a problem hiding this comment.
I have added a helper here to parse STREAMS_GROUP_RACK_AWARE_ASSIGNMENT_TAGS_DEFAULT
| // 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)), |
There was a problem hiding this comment.
Don't we need null check here and translate it to 0 ?
There was a problem hiding this 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
There was a problem hiding this 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.
| * 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(",")); |
There was a problem hiding this comment.
isEmpty() checks for length() == 0, right? What about " " (or similar). Do we need to to do rackAwareAssignmentTags.trim().isEmpty()` ?
There was a problem hiding this 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
3 participants
Adding AssignmentConfigs a public API for KIP-1357.
Reviewers: Sean Quah squah@confluent.io, Matthias J. Sax
matthias@confluent.io