KAFKA-18311: Internal Topic Manager (5/5) by lucasbru · Pull Request #18442 · apache/kafka

lucasbru · 2025-01-08T12:45:13Z

The internal topic manager takes a requested topology and returns a configured topology, as well as any internal topics that need to be created.

Shares with the client-side "InternalTopicManager" the name only.

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-08T12:46:09Z

PTAL @aliehsaeedii

lucasbru · 2025-01-10T15:09:24Z

@cadonna I moved the source topic check as you suggested.

cadonna

Thanks for the PR, @lucasbru !

Here my comments.

cadonna · 2025-01-13T09:05:48Z

+/**
+ * Immutable topic metadata, representing the current state of a topic in the broker.
+ *
+ * @param id             The topic id.


nit:

Suggested change

* @param id The topic id.

* @param id The topic ID.

cadonna · 2025-01-13T09:07:24Z

+    public TopicMetadata(
+        Uuid id,
+        String name,
+        int numPartitions,
+        Map<Integer, Set<String>> partitionRacks
+    ) {


The following is the code style we usually use in Streams and that is more readable IMO.

Suggested change

public TopicMetadata(

Uuid id,

String name,

int numPartitions,

Map<Integer, Set<String>> partitionRacks

) {

public TopicMetadata(Uuid id,

String name,

int numPartitions,

Map<Integer, Set<String>> partitionRacks) {

cadonna · 2025-01-13T09:10:33Z

+    public static TopicMetadata fromRecord(
+        StreamsGroupPartitionMetadataValue.TopicMetadata record
+    ) {


Suggested change

public static TopicMetadata fromRecord(

StreamsGroupPartitionMetadataValue.TopicMetadata record

) {

public static TopicMetadata fromRecord(StreamsGroupPartitionMetadataValue.TopicMetadata record) {

cadonna · 2025-01-13T09:19:49Z

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


Could you add a test with partition number 0?
This seems a special case that should be documented with a unit test.

Hmm, actually not sure why the implementation allows this. I changed it to require positive, let's see if it breaks anything.

cadonna · 2025-01-13T09:20:46Z

+public class TopicMetadataTest {
+
+    @Test
+    public void testConstructorWithZeroUuid() {


A positive constructor unit test is missing.

cadonna · 2025-01-13T09:39:40Z

+    private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology) {


Suggested change

private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(

final StreamsGroupTopologyValue.Subtopology subtopology) {

private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(final StreamsGroupTopologyValue.Subtopology subtopology) {

This will make the line too long, and my IDE (based on imported checkstyle config) will wrap it the same way.... I added another line break.

I didn't know this was an accepted standard in the streams code that everybody follows. Is there something I can do in my IDE to follow this? I imported the official Kafka checkstyle configuration, and it produces this kind of wrapping. It would be good if our "review nitpicking" and automated formatting tools somewhat agree.

I agree but I do also not know where it comes from. I would put it in one line although it exceeds the 120 characters, but it is also fine as you did it. The best would be to have an automatic formatter, but we never manged in the past to setup one. There was an attempt: #10428 but it was never merged.

cadonna · 2025-01-13T09:40:34Z

+    private static ConfiguredInternalTopic fromPersistedTopicInfo(
+        final StreamsGroupTopologyValue.TopicInfo topicInfo,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics) {


Suggested change

private static ConfiguredInternalTopic fromPersistedTopicInfo(

final StreamsGroupTopologyValue.TopicInfo topicInfo,

Map<String, Integer> decidedPartitionCountsForInternalTopics) {

private static ConfiguredInternalTopic fromPersistedTopicInfo(final StreamsGroupTopologyValue.TopicInfo topicInfo,

final Map<String, Integer> decidedPartitionCountsForInternalTopics) {

cadonna · 2025-01-13T09:52:34Z

+    private static ConfiguredSubtopology fromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics
+    ) {


Suggested change

private static ConfiguredSubtopology fromPersistedSubtopology(

final StreamsGroupTopologyValue.Subtopology subtopology,

Map<String, Integer> decidedPartitionCountsForInternalTopics

) {

private static ConfiguredSubtopology fromPersistedSubtopology(final StreamsGroupTopologyValue.Subtopology subtopology,

final Map<String, Integer> decidedPartitionCountsForInternalTopics

) {

cadonna · 2025-01-13T09:58:55Z

+        final Set<String> fixedRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() != 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+        final Set<String> flexibleRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() == 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+
+        if (fixedRepartitionTopics.isEmpty() && flexibleRepartitionTopics.isEmpty()) {
+            log.info("Skipping the repartition topic validation since there are no repartition topics.");
+        } else {
+            // ensure the co-partitioning topics within the group have the same number of partitions,
+            // and enforce the number of partitions for those repartition topics to be the same if they
+            // are co-partitioned as well.
+            for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+                for (Set<String> copartitionGroup : copartitionGroups) {
+                    decidedPartitionCountsForInternalTopics.putAll(
+                        copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));
+                }
+            }
+        }


Maybe export this block into a private method, so that we have each of repartition topics setup, co-partition enforcer, and changelog topcis setup in one line. Just a proposal.

lucasbru

Thanks for the review, @cadonna! Ready for re-review.

lucasbru · 2025-01-20T09:33:20Z

+    public TopicMetadata(
+        Uuid id,
+        String name,
+        int numPartitions,
+        Map<Integer, Set<String>> partitionRacks
+    ) {


lucasbru · 2025-01-20T09:33:23Z

+/**
+ * Immutable topic metadata, representing the current state of a topic in the broker.
+ *
+ * @param id             The topic id.


lucasbru · 2025-01-20T09:33:29Z

+    public static TopicMetadata fromRecord(
+        StreamsGroupPartitionMetadataValue.TopicMetadata record
+    ) {


lucasbru · 2025-01-20T09:35:55Z

+        final Set<String> fixedRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() != 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+        final Set<String> flexibleRepartitionTopics =
+            topology.subtopologies().values().stream().flatMap(x ->
+                x.repartitionSourceTopics().stream().filter(y -> y.partitions() == 0)
+            ).map(StreamsGroupTopologyValue.TopicInfo::name).collect(Collectors.toSet());
+
+        if (fixedRepartitionTopics.isEmpty() && flexibleRepartitionTopics.isEmpty()) {
+            log.info("Skipping the repartition topic validation since there are no repartition topics.");
+        } else {
+            // ensure the co-partitioning topics within the group have the same number of partitions,
+            // and enforce the number of partitions for those repartition topics to be the same if they
+            // are co-partitioned as well.
+            for (Collection<Set<String>> copartitionGroups : copartitionGroupsBySubtopology.values()) {
+                for (Set<String> copartitionGroup : copartitionGroups) {
+                    decidedPartitionCountsForInternalTopics.putAll(
+                        copartitionedTopicsEnforcer.enforce(copartitionGroup, fixedRepartitionTopics, flexibleRepartitionTopics));
+                }
+            }
+        }


lucasbru · 2025-01-20T09:37:16Z

+    private static ConfiguredSubtopology fromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics
+    ) {


lucasbru · 2025-01-20T09:37:23Z

+    private static ConfiguredInternalTopic fromPersistedTopicInfo(
+        final StreamsGroupTopologyValue.TopicInfo topicInfo,
+        Map<String, Integer> decidedPartitionCountsForInternalTopics) {


lucasbru · 2025-01-20T09:37:36Z

+    private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology) {


This will make the line too long, and my IDE (based on imported checkstyle config) will wrap it the same way.... I added another line break.

I didn't know this was an accepted standard in the streams code that everybody follows. Is there something I can do in my IDE to follow this? I imported the official Kafka checkstyle configuration, and it produces this kind of wrapping. It would be good if our "review nitpicking" and automated formatting tools somewhat agree.

lucasbru · 2025-01-20T09:44:29Z

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


Hmm, actually not sure why the implementation allows this. I changed it to require positive, let's see if it breaks anything.

lucasbru · 2025-01-20T09:48:56Z

+public class TopicMetadataTest {
+
+    @Test
+    public void testConstructorWithZeroUuid() {


cadonna

Thanks for the updates!

I had just one very minor comment. I leave it to you whether you want to change that or not.

cadonna · 2025-01-20T10:54:15Z

+        enforceCopartitioning(topology, copartitionGroupsBySubtopology, log,
+            decidedPartitionCountsForInternalTopics, copartitionedTopicsEnforcer);


nit:

Suggested change

enforceCopartitioning(topology, copartitionGroupsBySubtopology, log,

decidedPartitionCountsForInternalTopics, copartitionedTopicsEnforcer);

enforceCopartitioning(

topology,

copartitionGroupsBySubtopology,

log,

decidedPartitionCountsForInternalTopics,

copartitionedTopicsEnforcer

);

documented here: https://kafka.apache.org/coding-guide
There are always exceptions like log-messages.

cadonna · 2025-01-20T11:00:59Z

+    private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
+        final StreamsGroupTopologyValue.Subtopology subtopology) {


I agree but I do also not know where it comes from. I would put it in one line although it exceeds the 120 characters, but it is also fine as you did it. The best would be to have an automatic formatter, but we never manged in the past to setup one. There was an attempt: #10428 but it was never merged.

The internal topic manager takes a requested topology and returns a configured topology, as well as any internal topics that need to be created. Shares with the client-side "InternalTopicManager" the name only. Reviewers: Bruno Cadonna <cadonna@apache.org>

lucasbru changed the title ~~KAFKA-18311: Internal Topic Manager~~ KAFKA-18311: Internal Topic Manager (5/5) Jan 8, 2025

lucasbru requested review from bbejeck, cadonna and mjsax January 8, 2025 12:45

lucasbru force-pushed the kip1071merge/topic_config5 branch from 42f344d to 3ee997b Compare January 8, 2025 12:49

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

impl

3b195a0

lucasbru force-pushed the kip1071merge/topic_config5 branch from 3ee997b to 3b195a0 Compare January 10, 2025 14:14

cadonna reviewed Jan 13, 2025

View reviewed changes

fixes

2afb24a

lucasbru commented Jan 20, 2025

View reviewed changes

cadonna approved these changes Jan 20, 2025

View reviewed changes

fixes

dc87863

lucasbru merged commit 7d39ba1 into apache:trunk Jan 21, 2025

		private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
		final StreamsGroupTopologyValue.Subtopology subtopology) {

		enforceCopartitioning(topology, copartitionGroupsBySubtopology, log,
		decidedPartitionCountsForInternalTopics, copartitionedTopicsEnforcer);

Conversation

lucasbru commented Jan 8, 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 8, 2025

Uh oh!

lucasbru commented Jan 10, 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!

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 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!

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!

lucasbru commented Jan 8, 2025 •

edited

Loading