From 6d49d28bda454576a3dcbecd06d7ff50a6d6d44c Mon Sep 17 00:00:00 2001 From: Lucas Brutschy Date: Thu, 19 Dec 2024 10:45:16 +0100 Subject: [PATCH 1/3] impl --- .../topics/CopartitionedTopicsEnforcer.java | 185 ++++++++++++++ .../CopartitionedTopicsEnforcerTest.java | 237 ++++++++++++++++++ 2 files changed, 422 insertions(+) create mode 100644 group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcer.java create mode 100644 group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcerTest.java diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcer.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcer.java new file mode 100644 index 0000000000000..d30f28cf5228b --- /dev/null +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcer.java @@ -0,0 +1,185 @@ +/* + * 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.topics; + +import org.apache.kafka.common.errors.StreamsInvalidTopologyException; +import org.apache.kafka.common.utils.LogContext; + +import org.slf4j.Logger; + +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; +import java.util.OptionalInt; +import java.util.Set; +import java.util.TreeMap; +import java.util.function.Function; +import java.util.stream.Collectors; + +/** + * This class is responsible for enforcing the number of partitions in copartitioned topics. For each copartition group, it checks whether + * the number of partitions for all repartition topics is the same, and enforces copartitioning for repartition topics whose number of + * partitions is not enforced by the topology. + */ +public class CopartitionedTopicsEnforcer { + + private final Logger log; + private final Function topicPartitionCountProvider; + + /** + * The constructor for the class. + * + * @param logContext The context for emitting log messages. + * @param topicPartitionCountProvider Returns the number of partitions for a given topic, representing the current state of the broker. + */ + public CopartitionedTopicsEnforcer(final LogContext logContext, + final Function topicPartitionCountProvider) { + this.log = logContext.logger(getClass()); + this.topicPartitionCountProvider = topicPartitionCountProvider; + } + + /** + * Enforces the number of partitions for copartitioned topics. + * + * @param copartitionedTopics The set of copartitioned topics. + * @param repartitionTopicPartitionCounts A map from repartition topics to their determined partition count. + * @param enforcedRepartitionTopics The set of repartition topics whose partition count is enforced by the topology. + * @throws TopicConfigurationException If source topics are missing, or there are topics in copartitionTopics that are not copartitioned + * according to topicPartitionCountProvider are not co-partitioned. + * + * @return A map from all repartition topics to their partition counts + */ + public Map enforce(final Set copartitionedTopics, + final Map repartitionTopicPartitionCounts, + final Set enforcedRepartitionTopics) throws StreamsInvalidTopologyException { + if (copartitionedTopics.isEmpty()) { + return Collections.emptyMap(); + } + final Map returnedPartitionCounts = new HashMap<>(); + + final Map repartitionTopicConfigs = + copartitionedTopics.stream() + .filter(repartitionTopicPartitionCounts::containsKey) + .collect(Collectors.toMap(topic -> topic, repartitionTopicPartitionCounts::get)); + + final Map nonRepartitionTopicPartitions = + copartitionedTopics.stream().filter(topic -> !repartitionTopicPartitionCounts.containsKey(topic)) + .collect(Collectors.toMap(topic -> topic, topic -> { + final OptionalInt topicPartitionCount = topicPartitionCountProvider.apply(topic); + if (topicPartitionCount.isEmpty()) { + final String str = String.format("Following topics are missing: [%s]", topic); + log.error(str); + throw TopicConfigurationException.missingSourceTopics(str); + } else { + return topicPartitionCount.getAsInt(); + } + })); + + final int numPartitionsToUseForRepartitionTopics; + + if (copartitionedTopics.equals(repartitionTopicConfigs.keySet())) { + + // if there's at least one repartition topic with enforced number of partitions + // validate that they all have same number of partitions + if (!enforcedRepartitionTopics.isEmpty()) { + numPartitionsToUseForRepartitionTopics = validateAndGetNumOfPartitions( + repartitionTopicPartitionCounts, + enforcedRepartitionTopics + ); + } else { + // If all topics for this co-partition group are repartition topics, + // then set the number of partitions to be the maximum of the number of partitions. + numPartitionsToUseForRepartitionTopics = getMaxPartitions(repartitionTopicPartitionCounts); + } + } else { + // Otherwise, use the number of partitions from external topics (which must all be the same) + numPartitionsToUseForRepartitionTopics = getSamePartitions(nonRepartitionTopicPartitions); + } + + // coerce all the repartition topics to use the decided number of partitions. + for (final Entry repartitionTopic : repartitionTopicPartitionCounts.entrySet()) { + returnedPartitionCounts.put(repartitionTopic.getKey(), numPartitionsToUseForRepartitionTopics); + if (enforcedRepartitionTopics.contains(repartitionTopic.getKey()) + && repartitionTopic.getValue() != numPartitionsToUseForRepartitionTopics) { + final String msg = String.format("Number of partitions [%d] of repartition topic [%s] " + + "doesn't match number of partitions [%d] of the source topic.", + repartitionTopic.getValue(), + repartitionTopic.getKey(), + numPartitionsToUseForRepartitionTopics); + throw TopicConfigurationException.incorrectlyPartitionedTopics(msg); + } + } + + return returnedPartitionCounts; + } + + private int getEnforcedPartitionCount(final String topicName, final Map repartitionTopicConfigs) { + Integer partitions = repartitionTopicConfigs.get(topicName); + if (partitions != null) { + return partitions; + } else { + throw new StreamsInvalidTopologyException("Number of partitions is not set for topic: " + topicName); + } + } + + private int validateAndGetNumOfPartitions(final Map repartitionTopics, + final Collection enforcedTopics) { + final String firstTopicName = enforcedTopics.iterator().next(); + + final int firstNumberOfPartitionsOfInternalTopic = getEnforcedPartitionCount(firstTopicName, repartitionTopics); + + for (final String topicName : enforcedTopics) { + final int numberOfPartitions = getEnforcedPartitionCount(topicName, repartitionTopics); + + if (numberOfPartitions != firstNumberOfPartitionsOfInternalTopic) { + final String msg = String.format("Following topics do not have the same number of partitions: [%s]", + new TreeMap<>(repartitionTopics)); + throw TopicConfigurationException.incorrectlyPartitionedTopics(msg); + } + } + + return firstNumberOfPartitionsOfInternalTopic; + } + + private int getSamePartitions(final Map nonRepartitionTopicsInCopartitionGroup) { + final int partitions = nonRepartitionTopicsInCopartitionGroup.values().iterator().next(); + for (final Entry entry : nonRepartitionTopicsInCopartitionGroup.entrySet()) { + if (entry.getValue() != partitions) { + final TreeMap sorted = new TreeMap<>(nonRepartitionTopicsInCopartitionGroup); + throw TopicConfigurationException.incorrectlyPartitionedTopics( + String.format("Following topics do not have the same number of partitions: [%s]", sorted)); + } + } + return partitions; + } + + private int getMaxPartitions(final Map repartitionTopicsInCopartitionGroup) { + int maxPartitions = 0; + + for (final Integer numPartitions : repartitionTopicsInCopartitionGroup.values()) { + maxPartitions = Integer.max(maxPartitions, numPartitions); + } + if (maxPartitions == 0) { + throw new StreamsInvalidTopologyException("All topics in the copartition group had undefined partition number: " + + repartitionTopicsInCopartitionGroup.keySet()); + } + return maxPartitions; + } + +} diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcerTest.java new file mode 100644 index 0000000000000..a3d8686ceaaf3 --- /dev/null +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcerTest.java @@ -0,0 +1,237 @@ +/* + * 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.topics; + +import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse.Status; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.Utils; + +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.Map; +import java.util.OptionalInt; +import java.util.Set; +import java.util.TreeMap; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class CopartitionedTopicsEnforcerTest { + + private static final LogContext LOG_CONTEXT = new LogContext(); + + private static OptionalInt emptyTopicPartitionProvider(String topic) { + return OptionalInt.empty(); + } + + private static OptionalInt firstSecondTopicConsistent(String topic) { + if (topic.equals("first") || topic.equals("second")) { + return OptionalInt.of(2); + } + return OptionalInt.empty(); + } + + private static OptionalInt firstSecondTopicInconsistent(String topic) { + if (topic.equals("first")) { + return OptionalInt.of(2); + } + if (topic.equals("second")) { + return OptionalInt.of(1); + } + return OptionalInt.empty(); + } + + @Test + public void shouldThrowTopicConfigurationExceptionIfNoPartitionsFoundForCoPartitionedTopic() { + final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, + CopartitionedTopicsEnforcerTest::emptyTopicPartitionProvider); + TopicConfigurationException ex = assertThrows(TopicConfigurationException.class, () -> validator.enforce(Collections.singleton("topic"), + Collections.emptyMap(), Collections.emptySet())); + assertEquals(Status.MISSING_SOURCE_TOPICS, ex.status()); + assertEquals("Following topics are missing: [topic]", ex.getMessage()); + } + + @Test + public void shouldThrowTopicConfigurationExceptionIfPartitionCountsForCoPartitionedTopicsDontMatch() { + final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, + CopartitionedTopicsEnforcerTest::firstSecondTopicInconsistent); + TopicConfigurationException ex = assertThrows(TopicConfigurationException.class, () -> validator.enforce(Set.of("first", "second"), + Collections.emptyMap(), Collections.emptySet())); + assertEquals(Status.INCORRECTLY_PARTITIONED_TOPICS, ex.status()); + assertEquals("Following topics do not have the same number of partitions: " + + "[{first=2, second=1}]", ex.getMessage()); + } + + + @Test + public void shouldEnforceCopartitioningOnRepartitionTopics() { + final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, + CopartitionedTopicsEnforcerTest::firstSecondTopicConsistent); + final String repartitionTopic = "repartitioned"; + + Map result = + validator.enforce(Set.of("first", "second", repartitionTopic), + Map.of( + repartitionTopic, 10 + ), + Collections.emptySet() + ); + + assertEquals(Map.of(repartitionTopic, 2), result); + } + + + @Test + public void shouldSetNumPartitionsToMaximumPartitionsWhenAllTopicsAreRepartitionTopics() { + final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, + CopartitionedTopicsEnforcerTest::emptyTopicPartitionProvider); + final String one = "one"; + final String two = "two"; + final String three = "three"; + + Map result = validator.enforce(Set.of( + one, + two, + three + ), + Map.of( + one, 1, + two, 15, + three, 5 + ), + Collections.emptySet() + ); + + assertEquals(Map.of( + one, 15, + two, 15, + three, 15 + ), result); + } + + @Test + public void shouldThrowAnExceptionIfTopicInfosWithEnforcedNumOfPartitionsHaveDifferentNumOfPartitions() { + final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, + CopartitionedTopicsEnforcerTest::firstSecondTopicConsistent); + final String topic1 = "repartitioned-1"; + final String topic2 = "repartitioned-2"; + + final TopicConfigurationException ex = assertThrows( + TopicConfigurationException.class, + () -> validator.enforce(Set.of(topic1, topic2), + Utils.mkMap( + Utils.mkEntry(topic1, 10), + Utils.mkEntry(topic2, 5) + ), + Set.of(topic1, topic2) + ) + ); + + final TreeMap sorted = new TreeMap<>( + Utils.mkMap(Utils.mkEntry(topic1, 10), + Utils.mkEntry(topic2, 5)) + ); + assertEquals(Status.INCORRECTLY_PARTITIONED_TOPICS, ex.status()); + assertEquals(String.format( + "Following topics do not have the same number of partitions: " + + "[%s]", sorted), ex.getMessage()); + } + + @Test + public void shouldNotThrowAnExceptionWhenTopicInfosWithEnforcedNumOfPartitionsAreValid() { + final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, + CopartitionedTopicsEnforcerTest::firstSecondTopicConsistent); + final String topic1 = "repartitioned-1"; + final String topic2 = "repartitioned-2"; + + final Map enforced = validator.enforce(Set.of(topic1, topic2), + Utils.mkMap( + Utils.mkEntry(topic1, 10), + Utils.mkEntry(topic2, 10) + ), + Set.of(topic1, topic2) + ); + + assertEquals(Map.of( + topic1, 10, + topic2, 10 + ), enforced); + } + + @Test + public void shouldThrowAnExceptionWhenNumberOfPartitionsOfNonRepartitionTopicAndRepartitionTopicWithEnforcedNumOfPartitionsDoNotMatch() { + final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, + CopartitionedTopicsEnforcerTest::firstSecondTopicConsistent); + final String topic1 = "repartitioned-1"; + + final TopicConfigurationException ex = assertThrows( + TopicConfigurationException.class, + () -> validator.enforce(Set.of(topic1, "second"), + Utils.mkMap(Utils.mkEntry(topic1, 10)), + Set.of(topic1)) + ); + + assertEquals(Status.INCORRECTLY_PARTITIONED_TOPICS, ex.status()); + assertEquals(String.format("Number of partitions [%s] " + + "of repartition topic [%s] " + + "doesn't match number of partitions [%s] of the source topic.", + 10, topic1, 2), ex.getMessage()); + } + + @Test + public void shouldNotThrowAnExceptionWhenNumberOfPartitionsOfNonRepartitionTopicAndRepartitionTopicWithEnforcedNumOfPartitionsMatch() { + final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, + CopartitionedTopicsEnforcerTest::firstSecondTopicConsistent); + final String topic1 = "repartitioned-1"; + + final Map enforced = validator.enforce(Set.of(topic1, "second"), + Utils.mkMap(Utils.mkEntry(topic1, 2)), + Set.of(topic1)); + + assertEquals(Map.of( + topic1, 2 + ), enforced); + } + + @Test + public void shouldDeductNumberOfPartitionsFromRepartitionTopicWithEnforcedNumberOfPartitions() { + final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, + CopartitionedTopicsEnforcerTest::firstSecondTopicConsistent); + final String topic1 = "repartitioned-1"; + final String topic2 = "repartitioned-2"; + final String topic3 = "repartitioned-3"; + + final Map enforced = validator.enforce(Set.of(topic1, topic2), + Utils.mkMap( + Utils.mkEntry(topic1, 2), + Utils.mkEntry(topic2, 5), + Utils.mkEntry(topic3, 2) + ), + Set.of( + topic1, topic3 + ) + ); + + assertEquals(Map.of( + topic1, 2, + topic2, 2, + topic3, 2 + ), enforced); + } + +} \ No newline at end of file From c5ce4c779d752e1775e2d6225bf5927c5ae9c669 Mon Sep 17 00:00:00 2001 From: Lucas Brutschy Date: Wed, 8 Jan 2025 10:49:06 +0100 Subject: [PATCH 2/3] fixes --- .../topics/CopartitionedTopicsEnforcer.java | 52 ++-- .../CopartitionedTopicsEnforcerTest.java | 250 ++++++++++-------- 2 files changed, 165 insertions(+), 137 deletions(-) diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcer.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcer.java index d30f28cf5228b..468365d713e11 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcer.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcer.java @@ -46,7 +46,10 @@ public class CopartitionedTopicsEnforcer { * The constructor for the class. * * @param logContext The context for emitting log messages. - * @param topicPartitionCountProvider Returns the number of partitions for a given topic, representing the current state of the broker. + * @param topicPartitionCountProvider Returns the number of partitions for a given topic, representing the current state of the broker + * as well as any partition number decisions that have already been made. In particular, we expect + * the number of partitions for all repartition topics defined, even if they do not exist in the + * broker yet. */ public CopartitionedTopicsEnforcer(final LogContext logContext, final Function topicPartitionCountProvider) { @@ -57,26 +60,27 @@ public CopartitionedTopicsEnforcer(final LogContext logContext, /** * Enforces the number of partitions for copartitioned topics. * - * @param copartitionedTopics The set of copartitioned topics. - * @param repartitionTopicPartitionCounts A map from repartition topics to their determined partition count. - * @param enforcedRepartitionTopics The set of repartition topics whose partition count is enforced by the topology. + * @param copartitionedTopics The set of copartitioned topics (external source topics and repartition topics). + * @param fixedRepartitionTopics The set of repartition topics whose partition count is fixed by the topology. + * @param flexibleRepartitionTopics The set of repartition topics whose partition count is flexible, and can be changed. + * * @throws TopicConfigurationException If source topics are missing, or there are topics in copartitionTopics that are not copartitioned * according to topicPartitionCountProvider are not co-partitioned. * - * @return A map from all repartition topics to their partition counts + * @return A map from all repartition topics in copartitionedTopics to their updated partition counts. */ public Map enforce(final Set copartitionedTopics, - final Map repartitionTopicPartitionCounts, - final Set enforcedRepartitionTopics) throws StreamsInvalidTopologyException { + final Set fixedRepartitionTopics, + final Set flexibleRepartitionTopics) throws StreamsInvalidTopologyException { if (copartitionedTopics.isEmpty()) { return Collections.emptyMap(); } final Map returnedPartitionCounts = new HashMap<>(); - final Map repartitionTopicConfigs = + final Map repartitionTopicPartitionCounts = copartitionedTopics.stream() - .filter(repartitionTopicPartitionCounts::containsKey) - .collect(Collectors.toMap(topic -> topic, repartitionTopicPartitionCounts::get)); + .filter(x -> fixedRepartitionTopics.contains(x) || flexibleRepartitionTopics.contains(x)) + .collect(Collectors.toMap(topic -> topic, this::getPartitionCount)); final Map nonRepartitionTopicPartitions = copartitionedTopics.stream().filter(topic -> !repartitionTopicPartitionCounts.containsKey(topic)) @@ -93,14 +97,14 @@ public Map enforce(final Set copartitionedTopics, final int numPartitionsToUseForRepartitionTopics; - if (copartitionedTopics.equals(repartitionTopicConfigs.keySet())) { + if (copartitionedTopics.equals(repartitionTopicPartitionCounts.keySet())) { - // if there's at least one repartition topic with enforced number of partitions + // if there's at least one repartition topic with fixed number of partitions // validate that they all have same number of partitions - if (!enforcedRepartitionTopics.isEmpty()) { + if (!fixedRepartitionTopics.isEmpty()) { numPartitionsToUseForRepartitionTopics = validateAndGetNumOfPartitions( repartitionTopicPartitionCounts, - enforcedRepartitionTopics + fixedRepartitionTopics ); } else { // If all topics for this co-partition group are repartition topics, @@ -115,7 +119,7 @@ public Map enforce(final Set copartitionedTopics, // coerce all the repartition topics to use the decided number of partitions. for (final Entry repartitionTopic : repartitionTopicPartitionCounts.entrySet()) { returnedPartitionCounts.put(repartitionTopic.getKey(), numPartitionsToUseForRepartitionTopics); - if (enforcedRepartitionTopics.contains(repartitionTopic.getKey()) + if (fixedRepartitionTopics.contains(repartitionTopic.getKey()) && repartitionTopic.getValue() != numPartitionsToUseForRepartitionTopics) { final String msg = String.format("Number of partitions [%d] of repartition topic [%s] " + "doesn't match number of partitions [%d] of the source topic.", @@ -129,23 +133,23 @@ public Map enforce(final Set copartitionedTopics, return returnedPartitionCounts; } - private int getEnforcedPartitionCount(final String topicName, final Map repartitionTopicConfigs) { - Integer partitions = repartitionTopicConfigs.get(topicName); - if (partitions != null) { - return partitions; + private int getPartitionCount(final String topicName) { + OptionalInt partitions = topicPartitionCountProvider.apply(topicName); + if (partitions.isPresent()) { + return partitions.getAsInt(); } else { throw new StreamsInvalidTopologyException("Number of partitions is not set for topic: " + topicName); } } private int validateAndGetNumOfPartitions(final Map repartitionTopics, - final Collection enforcedTopics) { - final String firstTopicName = enforcedTopics.iterator().next(); + final Collection fixedRepartitionTopics) { + final String firstTopicName = fixedRepartitionTopics.iterator().next(); - final int firstNumberOfPartitionsOfInternalTopic = getEnforcedPartitionCount(firstTopicName, repartitionTopics); + final int firstNumberOfPartitionsOfInternalTopic = getPartitionCount(firstTopicName); - for (final String topicName : enforcedTopics) { - final int numberOfPartitions = getEnforcedPartitionCount(topicName, repartitionTopics); + for (final String topicName : fixedRepartitionTopics) { + final int numberOfPartitions = getPartitionCount(topicName); if (numberOfPartitions != firstNumberOfPartitionsOfInternalTopic) { final String msg = String.format("Following topics do not have the same number of partitions: [%s]", diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcerTest.java index a3d8686ceaaf3..c80a829b15459 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcerTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcerTest.java @@ -27,6 +27,7 @@ import java.util.OptionalInt; import java.util.Set; import java.util.TreeMap; +import java.util.function.Function; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -35,43 +36,45 @@ public class CopartitionedTopicsEnforcerTest { private static final LogContext LOG_CONTEXT = new LogContext(); - private static OptionalInt emptyTopicPartitionProvider(String topic) { - return OptionalInt.empty(); - } - - private static OptionalInt firstSecondTopicConsistent(String topic) { - if (topic.equals("first") || topic.equals("second")) { - return OptionalInt.of(2); - } - return OptionalInt.empty(); - } - - private static OptionalInt firstSecondTopicInconsistent(String topic) { - if (topic.equals("first")) { - return OptionalInt.of(2); - } - if (topic.equals("second")) { - return OptionalInt.of(1); - } - return OptionalInt.empty(); + private static Function topicPartitionProvider(Map topicPartitionCounts) { + return topic -> { + Integer a = topicPartitionCounts.get(topic); + return a == null ? OptionalInt.empty() : OptionalInt.of(a); + }; } @Test public void shouldThrowTopicConfigurationExceptionIfNoPartitionsFoundForCoPartitionedTopic() { - final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, - CopartitionedTopicsEnforcerTest::emptyTopicPartitionProvider); - TopicConfigurationException ex = assertThrows(TopicConfigurationException.class, () -> validator.enforce(Collections.singleton("topic"), - Collections.emptyMap(), Collections.emptySet())); + final String topic = "topic"; + final Map topicPartitionCounts = Collections.emptyMap(); + final CopartitionedTopicsEnforcer enforcer = + new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); + + final TopicConfigurationException ex = assertThrows(TopicConfigurationException.class, () -> + enforcer.enforce( + Set.of(topic), + Set.of(), + Set.of() + )); assertEquals(Status.MISSING_SOURCE_TOPICS, ex.status()); assertEquals("Following topics are missing: [topic]", ex.getMessage()); } @Test public void shouldThrowTopicConfigurationExceptionIfPartitionCountsForCoPartitionedTopicsDontMatch() { - final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, - CopartitionedTopicsEnforcerTest::firstSecondTopicInconsistent); - TopicConfigurationException ex = assertThrows(TopicConfigurationException.class, () -> validator.enforce(Set.of("first", "second"), - Collections.emptyMap(), Collections.emptySet())); + final String firstSourceTopic = "first"; + final String secondSourceTopic = "second"; + final Map topicPartitionCounts = Map.of(firstSourceTopic, 2, secondSourceTopic, 1); + final CopartitionedTopicsEnforcer enforcer = + new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); + + final TopicConfigurationException ex = assertThrows(TopicConfigurationException.class, () -> + enforcer.enforce( + Set.of(firstSourceTopic, secondSourceTopic), + Set.of(), + Set.of() + ) + ); assertEquals(Status.INCORRECTLY_PARTITIONED_TOPICS, ex.status()); assertEquals("Following topics do not have the same number of partitions: " + "[{first=2, second=1}]", ex.getMessage()); @@ -80,16 +83,22 @@ public void shouldThrowTopicConfigurationExceptionIfPartitionCountsForCoPartitio @Test public void shouldEnforceCopartitioningOnRepartitionTopics() { - final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, - CopartitionedTopicsEnforcerTest::firstSecondTopicConsistent); + final String firstSourceTopic = "first"; + final String secondSourceTopic = "second"; final String repartitionTopic = "repartitioned"; - - Map result = - validator.enforce(Set.of("first", "second", repartitionTopic), - Map.of( - repartitionTopic, 10 - ), - Collections.emptySet() + final Map topicPartitionCounts = Map.of( + firstSourceTopic, 2, + secondSourceTopic, 2, + repartitionTopic, 10 + ); + final CopartitionedTopicsEnforcer enforcer = + new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); + + final Map result = + enforcer.enforce( + Set.of(firstSourceTopic, secondSourceTopic, repartitionTopic), + Set.of(), + Set.of(repartitionTopic) ); assertEquals(Map.of(repartitionTopic, 2), result); @@ -98,53 +107,53 @@ public void shouldEnforceCopartitioningOnRepartitionTopics() { @Test public void shouldSetNumPartitionsToMaximumPartitionsWhenAllTopicsAreRepartitionTopics() { - final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, - CopartitionedTopicsEnforcerTest::emptyTopicPartitionProvider); - final String one = "one"; - final String two = "two"; - final String three = "three"; - - Map result = validator.enforce(Set.of( - one, - two, - three - ), - Map.of( - one, 1, - two, 15, - three, 5 - ), - Collections.emptySet() + final String repartitionTopic1 = "repartitionTopic1"; + final String repartitionTopic2 = "repartitionTopic2"; + final String repartitionTopic3 = "repartitionTopic3"; + final Map topicPartitionCounts = Map.of( + repartitionTopic1, 1, + repartitionTopic2, 15, + repartitionTopic3, 5 + ); + final CopartitionedTopicsEnforcer enforcer = + new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); + + final Map result = enforcer.enforce( + Set.of(repartitionTopic1, repartitionTopic2, repartitionTopic3), + Set.of(), + Set.of(repartitionTopic1, repartitionTopic2, repartitionTopic3) ); assertEquals(Map.of( - one, 15, - two, 15, - three, 15 + repartitionTopic1, 15, + repartitionTopic2, 15, + repartitionTopic3, 15 ), result); } @Test public void shouldThrowAnExceptionIfTopicInfosWithEnforcedNumOfPartitionsHaveDifferentNumOfPartitions() { - final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, - CopartitionedTopicsEnforcerTest::firstSecondTopicConsistent); - final String topic1 = "repartitioned-1"; - final String topic2 = "repartitioned-2"; + final String repartitionTopic1 = "repartitioned-1"; + final String repartitionTopic2 = "repartitioned-2"; + final Map topicPartitionCounts = Map.of( + repartitionTopic1, 10, + repartitionTopic2, 5 + ); + final CopartitionedTopicsEnforcer enforcer = + new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); final TopicConfigurationException ex = assertThrows( TopicConfigurationException.class, - () -> validator.enforce(Set.of(topic1, topic2), - Utils.mkMap( - Utils.mkEntry(topic1, 10), - Utils.mkEntry(topic2, 5) - ), - Set.of(topic1, topic2) + () -> enforcer.enforce( + Set.of(repartitionTopic1, repartitionTopic2), + Set.of(repartitionTopic1, repartitionTopic2), + Set.of() ) ); final TreeMap sorted = new TreeMap<>( - Utils.mkMap(Utils.mkEntry(topic1, 10), - Utils.mkEntry(topic2, 5)) + Utils.mkMap(Utils.mkEntry(repartitionTopic1, 10), + Utils.mkEntry(repartitionTopic2, 5)) ); assertEquals(Status.INCORRECTLY_PARTITIONED_TOPICS, ex.status()); assertEquals(String.format( @@ -154,83 +163,98 @@ public void shouldThrowAnExceptionIfTopicInfosWithEnforcedNumOfPartitionsHaveDif @Test public void shouldNotThrowAnExceptionWhenTopicInfosWithEnforcedNumOfPartitionsAreValid() { - final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, - CopartitionedTopicsEnforcerTest::firstSecondTopicConsistent); - final String topic1 = "repartitioned-1"; - final String topic2 = "repartitioned-2"; - - final Map enforced = validator.enforce(Set.of(topic1, topic2), - Utils.mkMap( - Utils.mkEntry(topic1, 10), - Utils.mkEntry(topic2, 10) - ), - Set.of(topic1, topic2) + final String repartitionTopic1 = "repartitioned-1"; + final String repartitionTopic2 = "repartitioned-2"; + final Map topicPartitionCounts = Map.of( + repartitionTopic1, 10, + repartitionTopic2, 10 + ); + final CopartitionedTopicsEnforcer enforcer = + new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); + + final Map enforced = enforcer.enforce( + Set.of(repartitionTopic1, repartitionTopic2), + Set.of(), + Set.of(repartitionTopic1, repartitionTopic2) ); assertEquals(Map.of( - topic1, 10, - topic2, 10 + repartitionTopic1, 10, + repartitionTopic2, 10 ), enforced); } @Test public void shouldThrowAnExceptionWhenNumberOfPartitionsOfNonRepartitionTopicAndRepartitionTopicWithEnforcedNumOfPartitionsDoNotMatch() { - final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, - CopartitionedTopicsEnforcerTest::firstSecondTopicConsistent); - final String topic1 = "repartitioned-1"; + final String repartitionTopic1 = "repartitioned-1"; + final String firstSourceTopic = "first"; + final Map topicPartitionCounts = Map.of( + repartitionTopic1, 10, + firstSourceTopic, 2 + ); + final CopartitionedTopicsEnforcer enforcer = + new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); final TopicConfigurationException ex = assertThrows( TopicConfigurationException.class, - () -> validator.enforce(Set.of(topic1, "second"), - Utils.mkMap(Utils.mkEntry(topic1, 10)), - Set.of(topic1)) + () -> enforcer.enforce( + Set.of(repartitionTopic1, firstSourceTopic), + Set.of(repartitionTopic1), + Set.of()) ); assertEquals(Status.INCORRECTLY_PARTITIONED_TOPICS, ex.status()); assertEquals(String.format("Number of partitions [%s] " + "of repartition topic [%s] " + "doesn't match number of partitions [%s] of the source topic.", - 10, topic1, 2), ex.getMessage()); + 10, repartitionTopic1, 2), ex.getMessage()); } @Test public void shouldNotThrowAnExceptionWhenNumberOfPartitionsOfNonRepartitionTopicAndRepartitionTopicWithEnforcedNumOfPartitionsMatch() { - final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, - CopartitionedTopicsEnforcerTest::firstSecondTopicConsistent); - final String topic1 = "repartitioned-1"; + final String repartitionTopic1 = "repartitioned-1"; + final String firstSourceTopic = "first"; + final Map topicPartitionCounts = Map.of( + repartitionTopic1, 2, + firstSourceTopic, 2 + ); + final CopartitionedTopicsEnforcer enforcer = + new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); - final Map enforced = validator.enforce(Set.of(topic1, "second"), - Utils.mkMap(Utils.mkEntry(topic1, 2)), - Set.of(topic1)); + final Map enforced = enforcer.enforce( + Set.of(repartitionTopic1, firstSourceTopic), + Set.of(), + Set.of(repartitionTopic1) + ); assertEquals(Map.of( - topic1, 2 + repartitionTopic1, 2 ), enforced); } @Test public void shouldDeductNumberOfPartitionsFromRepartitionTopicWithEnforcedNumberOfPartitions() { - final CopartitionedTopicsEnforcer validator = new CopartitionedTopicsEnforcer(LOG_CONTEXT, - CopartitionedTopicsEnforcerTest::firstSecondTopicConsistent); - final String topic1 = "repartitioned-1"; - final String topic2 = "repartitioned-2"; - final String topic3 = "repartitioned-3"; - - final Map enforced = validator.enforce(Set.of(topic1, topic2), - Utils.mkMap( - Utils.mkEntry(topic1, 2), - Utils.mkEntry(topic2, 5), - Utils.mkEntry(topic3, 2) - ), - Set.of( - topic1, topic3 - ) + final String repartitionTopic1 = "repartitioned-1"; + final String repartitionTopic2 = "repartitioned-2"; + final String repartitionTopic3 = "repartitioned-3"; + final Map topicPartitionCounts = Map.of( + repartitionTopic1, 2, + repartitionTopic2, 5, + repartitionTopic3, 2 + ); + final CopartitionedTopicsEnforcer enforcer = + new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); + + final Map enforced = enforcer.enforce( + Set.of(repartitionTopic1, repartitionTopic2, repartitionTopic3), + Set.of(repartitionTopic1, repartitionTopic3), + Set.of(repartitionTopic2) ); assertEquals(Map.of( - topic1, 2, - topic2, 2, - topic3, 2 + repartitionTopic1, 2, + repartitionTopic2, 2, + repartitionTopic3, 2 ), enforced); } From 7efa1eaac95123fea183fbbef725bde747d0e51a Mon Sep 17 00:00:00 2001 From: Lucas Brutschy Date: Thu, 9 Jan 2025 13:52:14 +0100 Subject: [PATCH 3/3] fixes --- .../topics/CopartitionedTopicsEnforcer.java | 5 +- .../CopartitionedTopicsEnforcerTest.java | 133 ++++++++---------- 2 files changed, 60 insertions(+), 78 deletions(-) diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcer.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcer.java index 468365d713e11..4bcc229ef3432 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcer.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcer.java @@ -61,11 +61,12 @@ public CopartitionedTopicsEnforcer(final LogContext logContext, * Enforces the number of partitions for copartitioned topics. * * @param copartitionedTopics The set of copartitioned topics (external source topics and repartition topics). - * @param fixedRepartitionTopics The set of repartition topics whose partition count is fixed by the topology. + * @param fixedRepartitionTopics The set of repartition topics whose partition count is fixed by the topology sent by the + * client (in particular, when the user uses `repartition` in the DSL). * @param flexibleRepartitionTopics The set of repartition topics whose partition count is flexible, and can be changed. * * @throws TopicConfigurationException If source topics are missing, or there are topics in copartitionTopics that are not copartitioned - * according to topicPartitionCountProvider are not co-partitioned. + * according to topicPartitionCountProvider. * * @return A map from all repartition topics in copartitionedTopics to their updated partition counts. */ diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcerTest.java index c80a829b15459..d2c466157f9c8 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcerTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/CopartitionedTopicsEnforcerTest.java @@ -18,7 +18,6 @@ import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse.Status; import org.apache.kafka.common.utils.LogContext; -import org.apache.kafka.common.utils.Utils; import org.junit.jupiter.api.Test; @@ -35,6 +34,11 @@ public class CopartitionedTopicsEnforcerTest { private static final LogContext LOG_CONTEXT = new LogContext(); + private static final String REPARTITION_TOPIC_1 = "repartitioned-1"; + private static final String REPARTITION_TOPIC_2 = "repartitioned-2"; + private static final String REPARTITION_TOPIC_3 = "repartitioned-3"; + private static final String SOURCE_TOPIC_1 = "source-1"; + private static final String SOURCE_TOPIC_2 = "source-2"; private static Function topicPartitionProvider(Map topicPartitionCounts) { return topic -> { @@ -45,99 +49,86 @@ private static Function topicPartitionProvider(Map topicPartitionCounts = Collections.emptyMap(); final CopartitionedTopicsEnforcer enforcer = new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); final TopicConfigurationException ex = assertThrows(TopicConfigurationException.class, () -> enforcer.enforce( - Set.of(topic), + Set.of(SOURCE_TOPIC_1), Set.of(), Set.of() )); assertEquals(Status.MISSING_SOURCE_TOPICS, ex.status()); - assertEquals("Following topics are missing: [topic]", ex.getMessage()); + assertEquals(String.format("Following topics are missing: [%s]", SOURCE_TOPIC_1), ex.getMessage()); } @Test public void shouldThrowTopicConfigurationExceptionIfPartitionCountsForCoPartitionedTopicsDontMatch() { - final String firstSourceTopic = "first"; - final String secondSourceTopic = "second"; - final Map topicPartitionCounts = Map.of(firstSourceTopic, 2, secondSourceTopic, 1); + final Map topicPartitionCounts = Map.of(SOURCE_TOPIC_1, 2, SOURCE_TOPIC_2, 1); final CopartitionedTopicsEnforcer enforcer = new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); final TopicConfigurationException ex = assertThrows(TopicConfigurationException.class, () -> enforcer.enforce( - Set.of(firstSourceTopic, secondSourceTopic), + Set.of(SOURCE_TOPIC_1, SOURCE_TOPIC_2), Set.of(), Set.of() ) ); assertEquals(Status.INCORRECTLY_PARTITIONED_TOPICS, ex.status()); - assertEquals("Following topics do not have the same number of partitions: " + - "[{first=2, second=1}]", ex.getMessage()); + assertEquals(String.format("Following topics do not have the same number of partitions: " + + "[{%s=2, %s=1}]", SOURCE_TOPIC_1, SOURCE_TOPIC_2), ex.getMessage()); } - @Test public void shouldEnforceCopartitioningOnRepartitionTopics() { - final String firstSourceTopic = "first"; - final String secondSourceTopic = "second"; - final String repartitionTopic = "repartitioned"; final Map topicPartitionCounts = Map.of( - firstSourceTopic, 2, - secondSourceTopic, 2, - repartitionTopic, 10 + SOURCE_TOPIC_1, 2, + SOURCE_TOPIC_2, 2, + REPARTITION_TOPIC_1, 10 ); final CopartitionedTopicsEnforcer enforcer = new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); final Map result = enforcer.enforce( - Set.of(firstSourceTopic, secondSourceTopic, repartitionTopic), + Set.of(SOURCE_TOPIC_1, SOURCE_TOPIC_2, REPARTITION_TOPIC_1), Set.of(), - Set.of(repartitionTopic) + Set.of(REPARTITION_TOPIC_1) ); - assertEquals(Map.of(repartitionTopic, 2), result); + assertEquals(Map.of(REPARTITION_TOPIC_1, 2), result); } - @Test public void shouldSetNumPartitionsToMaximumPartitionsWhenAllTopicsAreRepartitionTopics() { - final String repartitionTopic1 = "repartitionTopic1"; - final String repartitionTopic2 = "repartitionTopic2"; - final String repartitionTopic3 = "repartitionTopic3"; final Map topicPartitionCounts = Map.of( - repartitionTopic1, 1, - repartitionTopic2, 15, - repartitionTopic3, 5 + REPARTITION_TOPIC_1, 1, + REPARTITION_TOPIC_2, 15, + REPARTITION_TOPIC_3, 5 ); final CopartitionedTopicsEnforcer enforcer = new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); final Map result = enforcer.enforce( - Set.of(repartitionTopic1, repartitionTopic2, repartitionTopic3), + Set.of(REPARTITION_TOPIC_1, REPARTITION_TOPIC_2, REPARTITION_TOPIC_3), Set.of(), - Set.of(repartitionTopic1, repartitionTopic2, repartitionTopic3) + Set.of(REPARTITION_TOPIC_1, REPARTITION_TOPIC_2, REPARTITION_TOPIC_3) ); assertEquals(Map.of( - repartitionTopic1, 15, - repartitionTopic2, 15, - repartitionTopic3, 15 + REPARTITION_TOPIC_1, 15, + REPARTITION_TOPIC_2, 15, + REPARTITION_TOPIC_3, 15 ), result); } @Test public void shouldThrowAnExceptionIfTopicInfosWithEnforcedNumOfPartitionsHaveDifferentNumOfPartitions() { - final String repartitionTopic1 = "repartitioned-1"; - final String repartitionTopic2 = "repartitioned-2"; final Map topicPartitionCounts = Map.of( - repartitionTopic1, 10, - repartitionTopic2, 5 + REPARTITION_TOPIC_1, 10, + REPARTITION_TOPIC_2, 5 ); final CopartitionedTopicsEnforcer enforcer = new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); @@ -145,15 +136,14 @@ public void shouldThrowAnExceptionIfTopicInfosWithEnforcedNumOfPartitionsHaveDif final TopicConfigurationException ex = assertThrows( TopicConfigurationException.class, () -> enforcer.enforce( - Set.of(repartitionTopic1, repartitionTopic2), - Set.of(repartitionTopic1, repartitionTopic2), + Set.of(REPARTITION_TOPIC_1, REPARTITION_TOPIC_2), + Set.of(REPARTITION_TOPIC_1, REPARTITION_TOPIC_2), Set.of() ) ); final TreeMap sorted = new TreeMap<>( - Utils.mkMap(Utils.mkEntry(repartitionTopic1, 10), - Utils.mkEntry(repartitionTopic2, 5)) + Map.of(REPARTITION_TOPIC_1, 10, REPARTITION_TOPIC_2, 5) ); assertEquals(Status.INCORRECTLY_PARTITIONED_TOPICS, ex.status()); assertEquals(String.format( @@ -162,35 +152,31 @@ public void shouldThrowAnExceptionIfTopicInfosWithEnforcedNumOfPartitionsHaveDif } @Test - public void shouldNotThrowAnExceptionWhenTopicInfosWithEnforcedNumOfPartitionsAreValid() { - final String repartitionTopic1 = "repartitioned-1"; - final String repartitionTopic2 = "repartitioned-2"; + public void shouldReturnThePartitionCountsUnchangedWhenTopicInfosWithEnforcedNumOfPartitionsAreValid() { final Map topicPartitionCounts = Map.of( - repartitionTopic1, 10, - repartitionTopic2, 10 + REPARTITION_TOPIC_1, 10, + REPARTITION_TOPIC_2, 10 ); final CopartitionedTopicsEnforcer enforcer = new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); final Map enforced = enforcer.enforce( - Set.of(repartitionTopic1, repartitionTopic2), + Set.of(REPARTITION_TOPIC_1, REPARTITION_TOPIC_2), Set.of(), - Set.of(repartitionTopic1, repartitionTopic2) + Set.of(REPARTITION_TOPIC_1, REPARTITION_TOPIC_2) ); assertEquals(Map.of( - repartitionTopic1, 10, - repartitionTopic2, 10 + REPARTITION_TOPIC_1, 10, + REPARTITION_TOPIC_2, 10 ), enforced); } @Test public void shouldThrowAnExceptionWhenNumberOfPartitionsOfNonRepartitionTopicAndRepartitionTopicWithEnforcedNumOfPartitionsDoNotMatch() { - final String repartitionTopic1 = "repartitioned-1"; - final String firstSourceTopic = "first"; final Map topicPartitionCounts = Map.of( - repartitionTopic1, 10, - firstSourceTopic, 2 + REPARTITION_TOPIC_1, 10, + SOURCE_TOPIC_1, 2 ); final CopartitionedTopicsEnforcer enforcer = new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); @@ -198,8 +184,8 @@ public void shouldThrowAnExceptionWhenNumberOfPartitionsOfNonRepartitionTopicAnd final TopicConfigurationException ex = assertThrows( TopicConfigurationException.class, () -> enforcer.enforce( - Set.of(repartitionTopic1, firstSourceTopic), - Set.of(repartitionTopic1), + Set.of(REPARTITION_TOPIC_1, SOURCE_TOPIC_1), + Set.of(REPARTITION_TOPIC_1), Set.of()) ); @@ -207,54 +193,49 @@ public void shouldThrowAnExceptionWhenNumberOfPartitionsOfNonRepartitionTopicAnd assertEquals(String.format("Number of partitions [%s] " + "of repartition topic [%s] " + "doesn't match number of partitions [%s] of the source topic.", - 10, repartitionTopic1, 2), ex.getMessage()); + 10, REPARTITION_TOPIC_1, 2), ex.getMessage()); } @Test - public void shouldNotThrowAnExceptionWhenNumberOfPartitionsOfNonRepartitionTopicAndRepartitionTopicWithEnforcedNumOfPartitionsMatch() { - final String repartitionTopic1 = "repartitioned-1"; - final String firstSourceTopic = "first"; + public void shouldReturnThePartitionCountsUnchangedWhenNumberOfPartitionsOfNonRepartitionTopicAndRepartitionTopicWithEnforcedNumOfPartitionsMatch() { final Map topicPartitionCounts = Map.of( - repartitionTopic1, 2, - firstSourceTopic, 2 + REPARTITION_TOPIC_1, 2, + SOURCE_TOPIC_1, 2 ); final CopartitionedTopicsEnforcer enforcer = new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); final Map enforced = enforcer.enforce( - Set.of(repartitionTopic1, firstSourceTopic), + Set.of(REPARTITION_TOPIC_1, SOURCE_TOPIC_1), Set.of(), - Set.of(repartitionTopic1) + Set.of(REPARTITION_TOPIC_1) ); assertEquals(Map.of( - repartitionTopic1, 2 + REPARTITION_TOPIC_1, 2 ), enforced); } @Test public void shouldDeductNumberOfPartitionsFromRepartitionTopicWithEnforcedNumberOfPartitions() { - final String repartitionTopic1 = "repartitioned-1"; - final String repartitionTopic2 = "repartitioned-2"; - final String repartitionTopic3 = "repartitioned-3"; final Map topicPartitionCounts = Map.of( - repartitionTopic1, 2, - repartitionTopic2, 5, - repartitionTopic3, 2 + REPARTITION_TOPIC_1, 2, + REPARTITION_TOPIC_2, 5, + REPARTITION_TOPIC_3, 2 ); final CopartitionedTopicsEnforcer enforcer = new CopartitionedTopicsEnforcer(LOG_CONTEXT, topicPartitionProvider(topicPartitionCounts)); final Map enforced = enforcer.enforce( - Set.of(repartitionTopic1, repartitionTopic2, repartitionTopic3), - Set.of(repartitionTopic1, repartitionTopic3), - Set.of(repartitionTopic2) + Set.of(REPARTITION_TOPIC_1, REPARTITION_TOPIC_2, REPARTITION_TOPIC_3), + Set.of(REPARTITION_TOPIC_1, REPARTITION_TOPIC_3), + Set.of(REPARTITION_TOPIC_2) ); assertEquals(Map.of( - repartitionTopic1, 2, - repartitionTopic2, 2, - repartitionTopic3, 2 + REPARTITION_TOPIC_1, 2, + REPARTITION_TOPIC_2, 2, + REPARTITION_TOPIC_3, 2 ), enforced); }