Skip to content

KAFKA-18311: Internal Topic Manager (5/5) #18442

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lucasbru merged 3 commits into from
Jan 21, 2025
Merged

KAFKA-18311: Internal Topic Manager (5/5) #18442

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3b195a0
impl
lucasbru Jan 7, 2025
2afb24a
fixes
lucasbru Jan 20, 2025
dc87863
fixes
lucasbru Jan 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 8 additions & 12 deletions ...p-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TopicMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,20 +27,18 @@
/**
* Immutable topic metadata, representing the current state of a topic in the broker.
*
* @param id The topic id.
* @param id The topic ID.
* @param name The topic name.
* @param numPartitions The number of partitions.
* @param partitionRacks Map of every partition ID to a set of its rack IDs, if they exist. If rack information is unavailable for all
* partitions, this is an empty map.
*/
public record 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
) {
public TopicMetadata(Uuid id,
String name,
int numPartitions,
Map<Integer, Set<String>> partitionRacks) {
this.id = Objects.requireNonNull(id);
if (Uuid.ZERO_UUID.equals(id)) {
throw new IllegalArgumentException("Topic id cannot be ZERO_UUID.");
Expand All @@ -50,15 +48,13 @@ public TopicMetadata(
throw new IllegalArgumentException("Topic name cannot be empty.");
}
this.numPartitions = numPartitions;
if (numPartitions < 0) {
throw new IllegalArgumentException("Number of partitions cannot be negative.");
if (numPartitions <= 0) {
throw new IllegalArgumentException("Number of partitions must be positive.");
}
this.partitionRacks = Objects.requireNonNull(partitionRacks);
}

public static TopicMetadata fromRecord(
StreamsGroupPartitionMetadataValue.TopicMetadata record
) {
public static TopicMetadata fromRecord(StreamsGroupPartitionMetadataValue.TopicMetadata record) {
// Converting the data type from a list stored in the record to a map for the topic metadata.
Map<Integer, Set<String>> partitionRacks = new HashMap<>();
for (StreamsGroupPartitionMetadataValue.PartitionMetadata partitionMetadata : record.partitionMetadata()) {
Expand Down
30 changes: 19 additions & 11 deletions ...src/main/java/org/apache/kafka/coordinator/group/streams/topics/InternalTopicManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,6 +148,19 @@ private static Map<String, Integer> decidePartitionCounts(final LogContext logCo

decidedPartitionCountsForInternalTopics.putAll(repartitionTopics.setup());

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

@cadonna cadonna Jan 20, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@lucasbru lucasbru Jan 20, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done


decidedPartitionCountsForInternalTopics.putAll(changelogTopics.setup());

return decidedPartitionCountsForInternalTopics;
}

private static void enforceCopartitioning(final StreamsTopology topology,
final Map<String, Collection<Set<String>>> copartitionGroupsBySubtopology,
final Logger log,
final Map<String, Integer> decidedPartitionCountsForInternalTopics,
final CopartitionedTopicsEnforcer copartitionedTopicsEnforcer) {
final Set<String> fixedRepartitionTopics =
topology.subtopologies().values().stream().flatMap(x ->
x.repartitionSourceTopics().stream().filter(y -> y.partitions() != 0)
Expand All @@ -170,10 +183,6 @@ private static Map<String, Integer> decidePartitionCounts(final LogContext logCo
}
}
}
Comment on lines +169 to +190

@cadonna cadonna Jan 13, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done


decidedPartitionCountsForInternalTopics.putAll(changelogTopics.setup());

return decidedPartitionCountsForInternalTopics;
}

private static Map<String, CreatableTopic> missingInternalTopics(Map<String, ConfiguredSubtopology> subtopologyMap,
Expand Down Expand Up @@ -241,9 +250,8 @@ private static CreatableTopic toCreatableTopic(final ConfiguredInternalTopic con
return creatableTopic;
}

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
) {
return new ConfiguredSubtopology(
new HashSet<>(subtopology.sourceTopics()),
Expand All @@ -257,9 +265,8 @@ private static ConfiguredSubtopology fromPersistedSubtopology(
);
}

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) {
if (topicInfo.partitions() == 0 && !decidedPartitionCountsForInternalTopics.containsKey(topicInfo.name())) {
throw new IllegalStateException("Number of partitions must be set for topic " + topicInfo.name());
}
Expand All @@ -277,7 +284,8 @@ private static ConfiguredInternalTopic fromPersistedTopicInfo(
}

private static Collection<Set<String>> copartitionGroupsFromPersistedSubtopology(
final StreamsGroupTopologyValue.Subtopology subtopology) {
final StreamsGroupTopologyValue.Subtopology subtopology
) {
return subtopology.copartitionGroups().stream().map(copartitionGroup ->
Stream.concat(
copartitionGroup.sourceTopics().stream()
Expand Down
16 changes: 15 additions & 1 deletion ...ordinator/src/test/java/org/apache/kafka/coordinator/group/streams/TopicMetadataTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,11 +26,18 @@
import java.util.Map;
import java.util.Set;

import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;

public class TopicMetadataTest {

@Test
public void testConstructor() {
assertDoesNotThrow(() ->
new TopicMetadata(Uuid.randomUuid(), "valid-topic", 3, new HashMap<>()));
}

@Test
public void testConstructorWithZeroUuid() {

@cadonna cadonna Jan 13, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A positive constructor unit test is missing.

@lucasbru lucasbru Jan 20, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done

Exception exception = assertThrows(IllegalArgumentException.class, () ->
Expand All @@ -57,11 +64,18 @@ public void testConstructorWithEmptyName() {
assertEquals("Topic name cannot be empty.", exception.getMessage());
}

@Test
public void testConstructorWithZeroNumPartitions() {
Exception exception = assertThrows(IllegalArgumentException.class, () ->
new TopicMetadata(Uuid.randomUuid(), "valid-topic", 0, new HashMap<>()));
assertEquals("Number of partitions must be positive.", exception.getMessage());
}

@Test
public void testConstructorWithNegativeNumPartitions() {

@cadonna cadonna Jan 13, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@lucasbru lucasbru Jan 20, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Exception exception = assertThrows(IllegalArgumentException.class, () ->
new TopicMetadata(Uuid.randomUuid(), "valid-topic", -1, new HashMap<>()));
assertEquals("Number of partitions cannot be negative.", exception.getMessage());
assertEquals("Number of partitions must be positive.", exception.getMessage());
}

@Test
Expand Down