gabriellefu commented on code in PR #22213:
URL: https://github.com/apache/kafka/pull/22213#discussion_r3641014978
##########
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:
I have changed this, was considering List maybe but think it's a dead
branch. 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
