KAFKA-18311: Enforcing copartitioned topics (4/N) by lucasbru · Pull Request #18397 · apache/kafka

lucasbru · 2025-01-06T13:29:06Z

A simplified port of "CopartitionedTopicsEnforcer" from the client-side to the group coordinator.

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 topics in the group is the same, and enforces the right number of partitions for repartition topics whose number of partitions is not enforced by the topology.

Compared to the client-side version, the implementation uses immutable data structures, and returns the computed number of partitions instead of modifying mutable data structures and calling the admin client.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

lucasbru · 2025-01-06T13:30:41Z

PTAL @aliehsaeedii

cadonna

Thanks for the PR!

Here my comment!

cadonna · 2025-01-09T10:31:32Z

+     * @throws TopicConfigurationException If source topics are missing, or there are topics in copartitionTopics that are not copartitioned
+     *                                     according to topicPartitionCountProvider are not co-partitioned.


Something is wrong with this sentence. Should it be:

If source topics are missing, or there are topics in {@code copartitionTopics} that are not co-partitioned according to topicPartitionCountProvider.

?

cadonna · 2025-01-09T10:35:08Z

+     * 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.


With by the topology you mean user-specifed, right? If yes, maybe you should mention that they are user-specified.

I don't want too restrictive - the number of partitions is fixed in the topology, and on the broker we don't necessarily know why it was fixed. But I think right now you are right, this only happens when the user explicitly specifies it in repartition, so I updated it this way:

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

cadonna · 2025-01-09T10:38:59Z

+                    if (topicPartitionCount.isEmpty()) {
+                        final String str = String.format("Following topics are missing: [%s]", topic);
+                        log.error(str);
+                        throw TopicConfigurationException.missingSourceTopics(str);


Do you think, we are able to verify missing source topics in one location instead of spreading the verification between co-partition enforcer and repartition topic configuration?

Since we always call RepartitionTopics before calling this class in practice, this can never happen. But that it cannot happen, depends on how this class is used. I think it is still good style for this "unit" to have well-defined behavior if it is used in a context where a source topic is missing. Would you propose that this class just implicitly depends on the validation happening somewhere else?

cadonna · 2025-01-09T11:02:02Z

+        assertEquals(Map.of(repartitionTopic, 2), result);
+    }
+
+


Suggested change

cadonna · 2025-01-09T11:05:26Z

+            Utils.mkMap(Utils.mkEntry(repartitionTopic1, 10),
+                Utils.mkEntry(repartitionTopic2, 5))


Could you also use Map.of() here?

cadonna · 2025-01-09T11:09:02Z

+    }
+
+    @Test
+    public void shouldNotThrowAnExceptionWhenTopicInfosWithEnforcedNumOfPartitionsAreValid() {


nit:
Could you try to formulate this in a positive way?

cadonna · 2025-01-09T11:10:21Z

+    }
+
+    @Test
+    public void shouldNotThrowAnExceptionWhenNumberOfPartitionsOfNonRepartitionTopicAndRepartitionTopicWithEnforcedNumOfPartitionsMatch() {


Also here could you please try to formulate this positively?

cadonna · 2025-01-09T11:12:22Z

+        final String firstSourceTopic = "first";
+        final String secondSourceTopic = "second";


nit:
If you specify all this variables in all tests as constants with sensible names, it would reduce a bit the noise within the tests.

Interesting that you think this is better. Done

lucasbru · 2025-01-09T12:58:45Z

Ready for re-review @cadonna

lucasbru · 2025-01-10T11:08:14Z

@cadonna As discussed offline, I can add a third validation for source topics in a different place and throw IllegalStateException here. I will do these changes as I rebase InternalTopicManager, as this is the only place where I can put it. Can we please merge this PR, so I can continue?

cadonna

LGTM

A simplified port of "CopartitionedTopicsEnforcer" from the client-side to the group coordinator. 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 topics in the group is the same, and enforces the right number of partitions for repartition topics whose number of partitions is not enforced by the topology. Compared to the client-side version, the implementation uses immutable data structures, and returns the computed number of partitions instead of modifying mutable data structures and calling the admin client. Reviewers: Bruno Cadonna <cadonna@apache.org>

…emove-metadata-version-methods-for-versions-older-than-3.0 * apache-github/trunk: KAFKA-18340: Change Dockerfile to use log4j2 yaml instead log4j properties (apache#18378) MINOR: fix flaky RemoteLogManagerTest#testStopPartitionsWithDeletion (apache#18474) KAFKA-18311: Enforcing copartitioned topics (4/N) (apache#18397) KAFKA-18308; Update CoordinatorSerde (apache#18455) KAFKA-18440: Convert AuthorizationException to fatal error in AdminClient (apache#18435) KAFKA-17671: Create better documentation for transactions (apache#17454) KAFKA-18304; Introduce json converter generator (apache#18458) MINOR: Clean up classic group tests (apache#18473) KAFKA-18399 Remove ZooKeeper from KafkaApis (2/N): CONTROLLED_SHUTDOWN and ENVELOPE (apache#18422) MINOR: improve StreamThread periodic processing log (apache#18430)

A simplified port of "CopartitionedTopicsEnforcer" from the client-side to the group coordinator. 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 topics in the group is the same, and enforces the right number of partitions for repartition topics whose number of partitions is not enforced by the topology. Compared to the client-side version, the implementation uses immutable data structures, and returns the computed number of partitions instead of modifying mutable data structures and calling the admin client. Reviewers: Bruno Cadonna <cadonna@apache.org>

lucasbru requested a review from cadonna January 6, 2025 13:29

lucasbru requested review from bbejeck and mjsax January 6, 2025 13:30

impl

6d49d28

lucasbru force-pushed the kip1071merge/topic_config4 branch from a3a1fc8 to 6d49d28 Compare January 6, 2025 13:36

fixes

c5ce4c7

cadonna reviewed Jan 9, 2025

View reviewed changes

cadonna added streams core Kafka Broker KIP-1071 PRs related to KIP-1071 labels Jan 9, 2025

fixes

7efa1ea

cadonna approved these changes Jan 10, 2025

View reviewed changes

lucasbru merged commit b9a952d into apache:trunk Jan 10, 2025

		* @throws TopicConfigurationException If source topics are missing, or there are topics in copartitionTopics that are not copartitioned
		* according to topicPartitionCountProvider are not co-partitioned.

		Utils.mkMap(Utils.mkEntry(repartitionTopic1, 10),
		Utils.mkEntry(repartitionTopic2, 5))

		final String firstSourceTopic = "first";
		final String secondSourceTopic = "second";

Conversation

lucasbru commented Jan 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

lucasbru commented Jan 6, 2025

Uh oh!

cadonna left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lucasbru commented Jan 9, 2025

Uh oh!

lucasbru commented Jan 10, 2025

Uh oh!

cadonna left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

lucasbru commented Jan 6, 2025 •

edited

Loading