You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucasbru (via GitHub)" <gi...@apache.org> on 2024/12/19 09:57:53 UTC
[PR] KAFKA-18311: Add internal datastructure for configuring topologies (1/N) [kafka]
lucasbru opened a new pull request, #18268:
URL: https://github.com/apache/kafka/pull/18268
Clients in the Streams Rebalance Protocol send an "unconfigured" representation of the topology to the broker. That is, the number of input topics and (some) internal topics is not fixed, regular expressions are not resolved. The broker takes this description of the topology and, together with the current state of the topics on the broker, derives a `ConfiguredTopology`. The configured topology is what is being returned from `StreamsGroupDescribe`, and has all number of partitions defined, and regular expressions resolved. The configured topology also contains missing internal topics that need to be created, and potentially configuration errors, such as missing source topics.
In this change, we add the internal data structures for representing the configured topology. They differ in some details from the data structures used in the RPCs. Most importantly, they can be evolved independently of the public interface.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18311: Add internal datastructure for configuring topologies (1/N) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18268:
URL: https://github.com/apache/kafka/pull/18268#discussion_r1903972660
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredSubtopology.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.message.StreamsGroupDescribeResponseData;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Internal representation of a subtopology.
+ * <p>
+ * The subtopology is configured according to the number of partitions available in the source topics. It has regular expressions already
+ * resolved and defined exactly the information that is being used by streams groups assignment reconciliation.
+ * <p>
+ * Configured subtopologies may be recreated every time the input topics used by the subtopology are modified.
+ */
+public record ConfiguredSubtopology(Set<String> repartitionSinkTopics,
+ Set<String> sourceTopics,
+ Map<String, ConfiguredInternalTopic> repartitionSourceTopics,
+ Map<String, ConfiguredInternalTopic> stateChangelogTopics) {
Review Comment:
Done
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredInternalTopic.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.internals.Topic;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * ConfiguredInternalTopic captures the properties required for configuring the internal topics we create for change-logs and repartitioning
Review Comment:
Done
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18311: Add internal datastructure for configuring topologies (1/N) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #18268:
URL: https://github.com/apache/kafka/pull/18268#issuecomment-2573049651
Thanks for the review @cadonna ! test failures are unrelated
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18311: Add internal datastructure for configuring topologies (1/N) [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #18268:
URL: https://github.com/apache/kafka/pull/18268#discussion_r1903937620
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredSubtopology.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.message.StreamsGroupDescribeResponseData;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Internal representation of a subtopology.
+ * <p>
+ * The subtopology is configured according to the number of partitions available in the source topics. It has regular expressions already
+ * resolved and defined exactly the information that is being used by streams groups assignment reconciliation.
+ * <p>
+ * Configured subtopologies may be recreated every time the input topics used by the subtopology are modified.
+ */
+public record ConfiguredSubtopology(Set<String> repartitionSinkTopics,
+ Set<String> sourceTopics,
Review Comment:
Should we throw if source topics are `null` or empty?
Should we throw if one of the other parameters is `null`?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18311: Add internal datastructure for configuring topologies (1/N) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #18268:
URL: https://github.com/apache/kafka/pull/18268#issuecomment-2569509020
cc @bbejeck @mjsax if you have some cycles for reviewing
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18311: Add internal datastructure for configuring topologies (1/N) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18268:
URL: https://github.com/apache/kafka/pull/18268#discussion_r1903973493
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredInternalTopic.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.internals.Topic;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * ConfiguredInternalTopic captures the properties required for configuring the internal topics we create for change-logs and repartitioning
+ * etc.
+ * <p>
+ * It is derived from the topology sent by the client, and the current state of the topics inside the broker. If the topics on the broker
+ * changes, the internal topic may need to be reconfigured.
+ */
Review Comment:
Sure why not. I didn't add parameter descriptions because I didn't have much to say about them that isn't clear from the name. Anyway, added those.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredSubtopologyTest.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * 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.message.StreamsGroupDescribeResponseData;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ConfiguredSubtopologyTest {
+
+ @Test
+ public void testAsStreamsGroupDescribeSubtopology() {
+ String subtopologyId = "subtopology1";
+
+ Set<String> sourceTopics = new HashSet<>(Set.of("sourceTopic1", "sourceTopic2"));
+ Set<String> repartitionSinkTopics = new HashSet<>(Set.of("repartitionSinkTopic1", "repartitionSinkTopic2"));
+
+ ConfiguredInternalTopic internalTopicMock = mock(ConfiguredInternalTopic.class);
+ StreamsGroupDescribeResponseData.TopicInfo topicInfo = new StreamsGroupDescribeResponseData.TopicInfo();
+ when(internalTopicMock.asStreamsGroupDescribeTopicInfo()).thenReturn(topicInfo);
+
+ Map<String, ConfiguredInternalTopic> repartitionSourceTopics = Map.of("repartitionSourceTopic1", internalTopicMock);
+ Map<String, ConfiguredInternalTopic> stateChangelogTopics = Map.of("stateChangelogTopic1", internalTopicMock);
+
+ ConfiguredSubtopology configuredSubtopology = new ConfiguredSubtopology(
+ repartitionSinkTopics, sourceTopics, repartitionSourceTopics, stateChangelogTopics);
Review Comment:
Done
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredTopologyTest.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.message.CreateTopicsRequestData.CreatableTopic;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ConfiguredTopologyTest {
+
+ @Test
+ public void testIsReady() {
+ ConfiguredTopology readyTopology = new ConfiguredTopology(
+ 1, new HashMap<>(), new HashMap<>(), Optional.empty());
+ assertTrue(readyTopology.isReady());
+
+ ConfiguredTopology notReadyTopology = new ConfiguredTopology(
+ 1, new HashMap<>(), new HashMap<>(), Optional.of(TopicConfigurationException.missingSourceTopics("missing")));
+ assertFalse(notReadyTopology.isReady());
+ }
+
+ @Test
+ public void testAsStreamsGroupDescribeTopology() {
+ int topologyEpoch = 1;
Review Comment:
DOne
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18311: Add internal datastructure for configuring topologies (1/N) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #18268:
URL: https://github.com/apache/kafka/pull/18268#issuecomment-2567972764
@cadonna I converted all classes to record classes. This required a lot of refactoring in the follow-up code, but now the tests you were worried about are not necessary about, since equals / hashcode are provided by the record classes. Could you please re-review?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18311: Add internal datastructure for configuring topologies (1/N) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18268:
URL: https://github.com/apache/kafka/pull/18268#discussion_r1903972141
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredInternalTopic.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.internals.Topic;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * ConfiguredInternalTopic captures the properties required for configuring the internal topics we create for change-logs and repartitioning
+ * etc.
+ * <p>
+ * It is derived from the topology sent by the client, and the current state of the topics inside the broker. If the topics on the broker
+ * changes, the internal topic may need to be reconfigured.
+ */
+public record ConfiguredInternalTopic(String name,
+ Map<String, String> topicConfigs,
+ int numberOfPartitions,
+ Optional<Short> replicationFactor) {
Review Comment:
Good point, then it also agrees with the KIP. Updated.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18311: Add internal datastructure for configuring topologies (1/N) [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #18268:
URL: https://github.com/apache/kafka/pull/18268#discussion_r1893795554
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredInternalTopic.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.internals.Topic;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * ConfiguredInternalTopic captures the properties required for configuring the internal topics we create for change-logs and repartitioning
+ * etc.
+ * <p>
+ * It is derived from the topology sent by the client, and the current state of the topics inside the broker. If the topics on the broker
+ * changes, the internal topic may need to be reconfigured.
+ */
+public class ConfiguredInternalTopic {
+
+ private final String name;
+ private final Map<String, String> topicConfigs;
+ private final Optional<Short> replicationFactor;
+ private final boolean enforceNumberOfPartitions;
+ private Optional<Integer> numberOfPartitions;
+
+ public ConfiguredInternalTopic(final String name) {
+ this(name, Collections.emptyMap(), Optional.empty(), Optional.empty());
+ }
+
+ public ConfiguredInternalTopic(final String name,
+ final Map<String, String> topicConfigs) {
+ this(name, topicConfigs, Optional.empty(), Optional.empty());
+ }
+
+ public ConfiguredInternalTopic(final String name,
+ final Map<String, String> topicConfigs,
+ final Optional<Integer> numberOfPartitions,
+ final Optional<Short> replicationFactor) {
+ this.name = Objects.requireNonNull(name, "name can't be null");
+ Topic.validate(name);
+ numberOfPartitions.ifPresent(ConfiguredInternalTopic::validateNumberOfPartitions);
+ this.topicConfigs = Objects.requireNonNull(topicConfigs, "topicConfigs can't be null");
+ this.numberOfPartitions = numberOfPartitions;
+ this.replicationFactor = replicationFactor;
+ this.enforceNumberOfPartitions = numberOfPartitions.isPresent();
+ }
+
+ private static void validateNumberOfPartitions(final int numberOfPartitions) {
+ if (numberOfPartitions < 1) {
+ throw new IllegalArgumentException("Number of partitions must be at least 1.");
+ }
+ }
+
+ public Map<String, String> topicConfigs() {
+ return topicConfigs;
+ }
+
+ public boolean hasEnforcedNumberOfPartitions() {
+ return enforceNumberOfPartitions;
+ }
+
+ public String name() {
+ return name;
+ }
+
+ public Optional<Integer> numberOfPartitions() {
+ return numberOfPartitions;
+ }
+
+ public Optional<Short> replicationFactor() {
+ return replicationFactor;
+ }
+
+ public ConfiguredInternalTopic setNumberOfPartitions(final int numberOfPartitions) {
Review Comment:
As discussed offline, could you please change the visibility to package-private?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredInternalTopicTest.java:
##########
@@ -0,0 +1,125 @@
+/*
+ * 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.InvalidTopicException;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ConfiguredInternalTopicTest {
+
+ @Test
+ public void testConstructorAndGetters() {
+ Map<String, String> topicConfigs = new HashMap<>();
+ topicConfigs.put("retention.ms", "1000");
+
+ ConfiguredInternalTopic config = new ConfiguredInternalTopic("test-topic", topicConfigs, Optional.of(3), Optional.of((short) 2));
+
+ assertEquals("test-topic", config.name());
+ assertEquals(topicConfigs, config.topicConfigs());
+ assertEquals(Optional.of(3), config.numberOfPartitions());
+ assertEquals(Optional.of((short) 2), config.replicationFactor());
+ }
+
+ @Test
+ public void testConstructorWithNullName() {
+ assertThrows(NullPointerException.class, () -> new ConfiguredInternalTopic(null, Collections.emptyMap()));
+ }
+
+ @Test
+ public void testConstructorWithInvalidName() {
+ assertThrows(InvalidTopicException.class, () -> new ConfiguredInternalTopic("invalid topic name", Collections.emptyMap()));
+ }
+
+ @Test
+ public void testConstructorWithNullTopicConfigs() {
+ assertThrows(NullPointerException.class, () -> new ConfiguredInternalTopic("test-topic", null));
+ }
+
+ @Test
+ public void testSetNumberOfPartitions() {
+ ConfiguredInternalTopic config = new ConfiguredInternalTopic("test-topic", Collections.emptyMap());
+ config.setNumberOfPartitions(3);
+ assertEquals(Optional.of(3), config.numberOfPartitions());
+ }
+
+ @Test
+ public void testSetNumberOfPartitionsInvalid() {
+ ConfiguredInternalTopic config = new ConfiguredInternalTopic("test-topic", Collections.emptyMap());
+ assertThrows(IllegalArgumentException.class, () -> config.setNumberOfPartitions(0));
+ }
+
+ @Test
+ public void testSetNumberOfPartitionsUnsupportedOperation() {
+ ConfiguredInternalTopic config = new ConfiguredInternalTopic("test-topic", Collections.emptyMap(), Optional.of(3),
+ Optional.empty());
+ assertThrows(UnsupportedOperationException.class, () -> config.setNumberOfPartitions(4));
+ }
+
+ @Test
+ public void testEqualsAndHashCode() {
+ Map<String, String> topicConfigs = new HashMap<>();
+ topicConfigs.put("retention.ms", "1000");
+
+ ConfiguredInternalTopic config1 = new ConfiguredInternalTopic("test-topic", topicConfigs, Optional.of(3), Optional.of((short) 2));
+ ConfiguredInternalTopic config2 = new ConfiguredInternalTopic("test-topic", topicConfigs, Optional.of(3), Optional.of((short) 2));
+
+ assertEquals(config1, config2);
+ assertEquals(config1.hashCode(), config2.hashCode());
+ }
+
+ @Test
+ public void testToString() {
Review Comment:
nit:
Is this needed?
The `toString()` is mainly for debug purposes, right?
If you want to leave it, that is fine with me. Just asking.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredInternalTopicTest.java:
##########
@@ -0,0 +1,125 @@
+/*
+ * 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.InvalidTopicException;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ConfiguredInternalTopicTest {
+
+ @Test
+ public void testConstructorAndGetters() {
+ Map<String, String> topicConfigs = new HashMap<>();
+ topicConfigs.put("retention.ms", "1000");
+
+ ConfiguredInternalTopic config = new ConfiguredInternalTopic("test-topic", topicConfigs, Optional.of(3), Optional.of((short) 2));
+
+ assertEquals("test-topic", config.name());
+ assertEquals(topicConfigs, config.topicConfigs());
+ assertEquals(Optional.of(3), config.numberOfPartitions());
+ assertEquals(Optional.of((short) 2), config.replicationFactor());
+ }
+
+ @Test
+ public void testConstructorWithNullName() {
+ assertThrows(NullPointerException.class, () -> new ConfiguredInternalTopic(null, Collections.emptyMap()));
+ }
+
+ @Test
+ public void testConstructorWithInvalidName() {
+ assertThrows(InvalidTopicException.class, () -> new ConfiguredInternalTopic("invalid topic name", Collections.emptyMap()));
+ }
+
+ @Test
+ public void testConstructorWithNullTopicConfigs() {
+ assertThrows(NullPointerException.class, () -> new ConfiguredInternalTopic("test-topic", null));
+ }
+
+ @Test
+ public void testSetNumberOfPartitions() {
+ ConfiguredInternalTopic config = new ConfiguredInternalTopic("test-topic", Collections.emptyMap());
+ config.setNumberOfPartitions(3);
+ assertEquals(Optional.of(3), config.numberOfPartitions());
+ }
+
+ @Test
+ public void testSetNumberOfPartitionsInvalid() {
+ ConfiguredInternalTopic config = new ConfiguredInternalTopic("test-topic", Collections.emptyMap());
+ assertThrows(IllegalArgumentException.class, () -> config.setNumberOfPartitions(0));
+ }
+
+ @Test
+ public void testSetNumberOfPartitionsUnsupportedOperation() {
+ ConfiguredInternalTopic config = new ConfiguredInternalTopic("test-topic", Collections.emptyMap(), Optional.of(3),
+ Optional.empty());
+ assertThrows(UnsupportedOperationException.class, () -> config.setNumberOfPartitions(4));
+ }
+
+ @Test
+ public void testEqualsAndHashCode() {
+ Map<String, String> topicConfigs = new HashMap<>();
+ topicConfigs.put("retention.ms", "1000");
+
+ ConfiguredInternalTopic config1 = new ConfiguredInternalTopic("test-topic", topicConfigs, Optional.of(3), Optional.of((short) 2));
+ ConfiguredInternalTopic config2 = new ConfiguredInternalTopic("test-topic", topicConfigs, Optional.of(3), Optional.of((short) 2));
+
+ assertEquals(config1, config2);
+ assertEquals(config1.hashCode(), config2.hashCode());
Review Comment:
Could you please also add a negative test for each field that is triggers an inequality?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredSubtopologyTest.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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.message.StreamsGroupDescribeResponseData;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ConfiguredSubtopologyTest {
+
+ @Test
+ public void testConstructorAndGetters() {
+ Set<String> repartitionSinkTopics = Set.of("repartitionSinkTopic1", "repartitionSinkTopic2");
+ Set<String> sourceTopics = Set.of("sourceTopic1", "sourceTopic2");
+ Map<String, ConfiguredInternalTopic> repartitionSourceTopics = new HashMap<>();
+ Map<String, ConfiguredInternalTopic> stateChangelogTopics = new HashMap<>();
+
+ ConfiguredSubtopology configuredSubtopology = new ConfiguredSubtopology(repartitionSinkTopics, sourceTopics,
+ repartitionSourceTopics, stateChangelogTopics);
+
+ assertEquals(repartitionSinkTopics, configuredSubtopology.repartitionSinkTopics());
+ assertEquals(sourceTopics, configuredSubtopology.sourceTopics());
+ assertEquals(repartitionSourceTopics, configuredSubtopology.repartitionSourceTopics());
+ assertEquals(stateChangelogTopics, configuredSubtopology.stateChangelogTopics());
+ }
+
+ @Test
+ public void testSetters() {
+ ConfiguredSubtopology configuredSubtopology = new ConfiguredSubtopology();
+
+ Set<String> repartitionSinkTopics = Set.of("repartitionSinkTopic1", "repartitionSinkTopic2");
+ ConfiguredSubtopology ret1 = configuredSubtopology.setRepartitionSinkTopics(repartitionSinkTopics);
+ assertEquals(repartitionSinkTopics, configuredSubtopology.repartitionSinkTopics());
+ assertEquals(ret1, configuredSubtopology);
+
+ Set<String> sourceTopics = Set.of("sourceTopic1", "sourceTopic2");
+ ConfiguredSubtopology ret2 = configuredSubtopology.setSourceTopics(sourceTopics);
+ assertEquals(sourceTopics, configuredSubtopology.sourceTopics());
+ assertEquals(ret2, configuredSubtopology);
+
+ Map<String, ConfiguredInternalTopic> repartitionSourceTopics = new HashMap<>();
+ ConfiguredSubtopology ret3 = configuredSubtopology.setRepartitionSourceTopics(repartitionSourceTopics);
+ assertEquals(repartitionSourceTopics, configuredSubtopology.repartitionSourceTopics());
+ assertEquals(ret3, configuredSubtopology);
+
+ Map<String, ConfiguredInternalTopic> stateChangelogTopics = new HashMap<>();
+ ConfiguredSubtopology ret4 = configuredSubtopology.setStateChangelogTopics(stateChangelogTopics);
+ assertEquals(stateChangelogTopics, configuredSubtopology.stateChangelogTopics());
+ assertEquals(ret4, configuredSubtopology);
+ }
+
+ @Test
+ public void testNonSourceChangelogTopics() {
+ Map<String, ConfiguredInternalTopic> stateChangelogTopics = new HashMap<>();
+ stateChangelogTopics.put("changelogTopic1", new ConfiguredInternalTopic("changelogTopic1"));
+ stateChangelogTopics.put("sourceTopic1", new ConfiguredInternalTopic("sourceTopic1"));
+
+ ConfiguredSubtopology configuredSubtopology = new ConfiguredSubtopology(
+ Collections.emptySet(),
+ Collections.singleton("sourceTopic1"),
+ Collections.emptyMap(),
+ stateChangelogTopics
+ );
+
+ Set<ConfiguredInternalTopic> nonSourceChangelogTopics = configuredSubtopology.nonSourceChangelogTopics();
+ assertEquals(1, nonSourceChangelogTopics.size());
+ assertTrue(nonSourceChangelogTopics.contains(new ConfiguredInternalTopic("changelogTopic1")));
Review Comment:
Could you please add a negative example?
That is, a source changelog topic that is not returned by `nonSourceChangelogTopics()`.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredSubtopologyTest.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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.message.StreamsGroupDescribeResponseData;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ConfiguredSubtopologyTest {
+
+ @Test
+ public void testConstructorAndGetters() {
+ Set<String> repartitionSinkTopics = Set.of("repartitionSinkTopic1", "repartitionSinkTopic2");
+ Set<String> sourceTopics = Set.of("sourceTopic1", "sourceTopic2");
+ Map<String, ConfiguredInternalTopic> repartitionSourceTopics = new HashMap<>();
+ Map<String, ConfiguredInternalTopic> stateChangelogTopics = new HashMap<>();
+
+ ConfiguredSubtopology configuredSubtopology = new ConfiguredSubtopology(repartitionSinkTopics, sourceTopics,
+ repartitionSourceTopics, stateChangelogTopics);
+
+ assertEquals(repartitionSinkTopics, configuredSubtopology.repartitionSinkTopics());
+ assertEquals(sourceTopics, configuredSubtopology.sourceTopics());
+ assertEquals(repartitionSourceTopics, configuredSubtopology.repartitionSourceTopics());
+ assertEquals(stateChangelogTopics, configuredSubtopology.stateChangelogTopics());
+ }
+
+ @Test
+ public void testSetters() {
+ ConfiguredSubtopology configuredSubtopology = new ConfiguredSubtopology();
+
+ Set<String> repartitionSinkTopics = Set.of("repartitionSinkTopic1", "repartitionSinkTopic2");
+ ConfiguredSubtopology ret1 = configuredSubtopology.setRepartitionSinkTopics(repartitionSinkTopics);
+ assertEquals(repartitionSinkTopics, configuredSubtopology.repartitionSinkTopics());
+ assertEquals(ret1, configuredSubtopology);
+
+ Set<String> sourceTopics = Set.of("sourceTopic1", "sourceTopic2");
+ ConfiguredSubtopology ret2 = configuredSubtopology.setSourceTopics(sourceTopics);
+ assertEquals(sourceTopics, configuredSubtopology.sourceTopics());
+ assertEquals(ret2, configuredSubtopology);
+
+ Map<String, ConfiguredInternalTopic> repartitionSourceTopics = new HashMap<>();
+ ConfiguredSubtopology ret3 = configuredSubtopology.setRepartitionSourceTopics(repartitionSourceTopics);
+ assertEquals(repartitionSourceTopics, configuredSubtopology.repartitionSourceTopics());
+ assertEquals(ret3, configuredSubtopology);
+
+ Map<String, ConfiguredInternalTopic> stateChangelogTopics = new HashMap<>();
+ ConfiguredSubtopology ret4 = configuredSubtopology.setStateChangelogTopics(stateChangelogTopics);
+ assertEquals(stateChangelogTopics, configuredSubtopology.stateChangelogTopics());
+ assertEquals(ret4, configuredSubtopology);
+ }
+
+ @Test
+ public void testNonSourceChangelogTopics() {
+ Map<String, ConfiguredInternalTopic> stateChangelogTopics = new HashMap<>();
+ stateChangelogTopics.put("changelogTopic1", new ConfiguredInternalTopic("changelogTopic1"));
+ stateChangelogTopics.put("sourceTopic1", new ConfiguredInternalTopic("sourceTopic1"));
+
+ ConfiguredSubtopology configuredSubtopology = new ConfiguredSubtopology(
+ Collections.emptySet(),
+ Collections.singleton("sourceTopic1"),
+ Collections.emptyMap(),
+ stateChangelogTopics
+ );
+
+ Set<ConfiguredInternalTopic> nonSourceChangelogTopics = configuredSubtopology.nonSourceChangelogTopics();
+ assertEquals(1, nonSourceChangelogTopics.size());
+ assertTrue(nonSourceChangelogTopics.contains(new ConfiguredInternalTopic("changelogTopic1")));
+ }
+
+ @Test
+ public void testEquals() {
+ Set<String> repartitionSinkTopics = new HashSet<>(Arrays.asList("repartitionSinkTopic1", "repartitionSinkTopic2"));
+ Set<String> sourceTopics = new HashSet<>(Arrays.asList("sourceTopic1", "sourceTopic2"));
+ Map<String, ConfiguredInternalTopic> repartitionSourceTopics = new HashMap<>();
+ Map<String, ConfiguredInternalTopic> stateChangelogTopics = new HashMap<>();
+
+ ConfiguredSubtopology configuredSubtopology1 = new ConfiguredSubtopology(repartitionSinkTopics, sourceTopics,
+ repartitionSourceTopics, stateChangelogTopics);
+ ConfiguredSubtopology configuredSubtopology2 = new ConfiguredSubtopology(repartitionSinkTopics, sourceTopics,
+ repartitionSourceTopics, stateChangelogTopics);
+
+ assertEquals(configuredSubtopology1, configuredSubtopology2);
+ }
+
+ @Test
+ public void testHashCode() {
Review Comment:
Consider merging this test with `testEquals()`. Might save some code line. Do not forget the neg. examples.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredInternalTopicTest.java:
##########
@@ -0,0 +1,125 @@
+/*
+ * 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.InvalidTopicException;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ConfiguredInternalTopicTest {
+
+ @Test
+ public void testConstructorAndGetters() {
+ Map<String, String> topicConfigs = new HashMap<>();
+ topicConfigs.put("retention.ms", "1000");
+
+ ConfiguredInternalTopic config = new ConfiguredInternalTopic("test-topic", topicConfigs, Optional.of(3), Optional.of((short) 2));
+
+ assertEquals("test-topic", config.name());
+ assertEquals(topicConfigs, config.topicConfigs());
+ assertEquals(Optional.of(3), config.numberOfPartitions());
+ assertEquals(Optional.of((short) 2), config.replicationFactor());
+ }
Review Comment:
nit:
Do you think this is needed?
I am fine if you want to keep it. I usually do not test this rather simple constructor and getters.
This also applies to the other test classes.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/TopicConfigurationException.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * 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;
+import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse.Status;
+
+import java.util.Objects;
+
+public class TopicConfigurationException extends RuntimeException {
+
+ private final Status status;
+
+ public TopicConfigurationException(StreamsGroupHeartbeatResponse.Status status, String message) {
+ super(message);
+ this.status = status;
+ }
+
+ public Status status() {
+ return status;
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final TopicConfigurationException that = (TopicConfigurationException) o;
+ return status == that.status && Objects.equals(getMessage(), that.getMessage());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(status, getMessage());
+ }
+
+ @Override
+ public String toString() {
+ return "TopicConfigurationException{" +
+ "status=" + status +
+ ", getMessage='" + getMessage() + '\'' +
+ '}';
+ }
Review Comment:
Do we need those overrides?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredSubtopologyTest.java:
##########
@@ -0,0 +1,172 @@
+/*
+ * 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.message.StreamsGroupDescribeResponseData;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ConfiguredSubtopologyTest {
+
+ @Test
+ public void testConstructorAndGetters() {
+ Set<String> repartitionSinkTopics = Set.of("repartitionSinkTopic1", "repartitionSinkTopic2");
+ Set<String> sourceTopics = Set.of("sourceTopic1", "sourceTopic2");
+ Map<String, ConfiguredInternalTopic> repartitionSourceTopics = new HashMap<>();
+ Map<String, ConfiguredInternalTopic> stateChangelogTopics = new HashMap<>();
+
+ ConfiguredSubtopology configuredSubtopology = new ConfiguredSubtopology(repartitionSinkTopics, sourceTopics,
+ repartitionSourceTopics, stateChangelogTopics);
+
+ assertEquals(repartitionSinkTopics, configuredSubtopology.repartitionSinkTopics());
+ assertEquals(sourceTopics, configuredSubtopology.sourceTopics());
+ assertEquals(repartitionSourceTopics, configuredSubtopology.repartitionSourceTopics());
+ assertEquals(stateChangelogTopics, configuredSubtopology.stateChangelogTopics());
+ }
+
+ @Test
+ public void testSetters() {
+ ConfiguredSubtopology configuredSubtopology = new ConfiguredSubtopology();
+
+ Set<String> repartitionSinkTopics = Set.of("repartitionSinkTopic1", "repartitionSinkTopic2");
+ ConfiguredSubtopology ret1 = configuredSubtopology.setRepartitionSinkTopics(repartitionSinkTopics);
+ assertEquals(repartitionSinkTopics, configuredSubtopology.repartitionSinkTopics());
+ assertEquals(ret1, configuredSubtopology);
+
+ Set<String> sourceTopics = Set.of("sourceTopic1", "sourceTopic2");
+ ConfiguredSubtopology ret2 = configuredSubtopology.setSourceTopics(sourceTopics);
+ assertEquals(sourceTopics, configuredSubtopology.sourceTopics());
+ assertEquals(ret2, configuredSubtopology);
+
+ Map<String, ConfiguredInternalTopic> repartitionSourceTopics = new HashMap<>();
+ ConfiguredSubtopology ret3 = configuredSubtopology.setRepartitionSourceTopics(repartitionSourceTopics);
+ assertEquals(repartitionSourceTopics, configuredSubtopology.repartitionSourceTopics());
+ assertEquals(ret3, configuredSubtopology);
+
+ Map<String, ConfiguredInternalTopic> stateChangelogTopics = new HashMap<>();
+ ConfiguredSubtopology ret4 = configuredSubtopology.setStateChangelogTopics(stateChangelogTopics);
+ assertEquals(stateChangelogTopics, configuredSubtopology.stateChangelogTopics());
+ assertEquals(ret4, configuredSubtopology);
+ }
+
+ @Test
+ public void testNonSourceChangelogTopics() {
+ Map<String, ConfiguredInternalTopic> stateChangelogTopics = new HashMap<>();
+ stateChangelogTopics.put("changelogTopic1", new ConfiguredInternalTopic("changelogTopic1"));
+ stateChangelogTopics.put("sourceTopic1", new ConfiguredInternalTopic("sourceTopic1"));
+
+ ConfiguredSubtopology configuredSubtopology = new ConfiguredSubtopology(
+ Collections.emptySet(),
+ Collections.singleton("sourceTopic1"),
+ Collections.emptyMap(),
+ stateChangelogTopics
+ );
+
+ Set<ConfiguredInternalTopic> nonSourceChangelogTopics = configuredSubtopology.nonSourceChangelogTopics();
+ assertEquals(1, nonSourceChangelogTopics.size());
+ assertTrue(nonSourceChangelogTopics.contains(new ConfiguredInternalTopic("changelogTopic1")));
+ }
+
+ @Test
+ public void testEquals() {
+ Set<String> repartitionSinkTopics = new HashSet<>(Arrays.asList("repartitionSinkTopic1", "repartitionSinkTopic2"));
+ Set<String> sourceTopics = new HashSet<>(Arrays.asList("sourceTopic1", "sourceTopic2"));
+ Map<String, ConfiguredInternalTopic> repartitionSourceTopics = new HashMap<>();
+ Map<String, ConfiguredInternalTopic> stateChangelogTopics = new HashMap<>();
+
+ ConfiguredSubtopology configuredSubtopology1 = new ConfiguredSubtopology(repartitionSinkTopics, sourceTopics,
+ repartitionSourceTopics, stateChangelogTopics);
+ ConfiguredSubtopology configuredSubtopology2 = new ConfiguredSubtopology(repartitionSinkTopics, sourceTopics,
+ repartitionSourceTopics, stateChangelogTopics);
+
+ assertEquals(configuredSubtopology1, configuredSubtopology2);
Review Comment:
Could you please add neg. examples?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredTopologyTest.java:
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.message.CreateTopicsRequestData.CreatableTopic;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ConfiguredTopologyTest {
+
+ @Test
+ public void testConstructorAndGetters() {
+ int topologyEpoch = 1;
+ Map<String, ConfiguredSubtopology> subtopologies = new HashMap<>();
+ Map<String, CreatableTopic> internalTopicsToBeCreated = new HashMap<>();
+ Optional<TopicConfigurationException> topicConfigurationException = Optional.empty();
+
+ ConfiguredTopology configuredTopology = new ConfiguredTopology(
+ topologyEpoch, subtopologies, internalTopicsToBeCreated, topicConfigurationException);
+
+ assertEquals(topologyEpoch, configuredTopology.topologyEpoch());
+ assertEquals(subtopologies, configuredTopology.subtopologies());
+ assertEquals(internalTopicsToBeCreated, configuredTopology.internalTopicsToBeCreated());
+ assertEquals(topicConfigurationException, configuredTopology.topicConfigurationException());
+ }
+
+ @Test
+ public void testIsReady() {
+ ConfiguredTopology readyTopology = new ConfiguredTopology(
+ 1, new HashMap<>(), new HashMap<>(), Optional.empty());
+ assertTrue(readyTopology.isReady());
+
+ ConfiguredTopology notReadyTopology = new ConfiguredTopology(
+ 1, new HashMap<>(), new HashMap<>(), Optional.of(TopicConfigurationException.missingSourceTopics("missing")));
+ assertFalse(notReadyTopology.isReady());
+ }
+
+ @Test
+ public void testEqualsAndHashCode() {
+ int topologyEpoch = 1;
+ Map<String, ConfiguredSubtopology> subtopologies = new HashMap<>();
+ Map<String, CreatableTopic> internalTopicsToBeCreated = new HashMap<>();
+ Optional<TopicConfigurationException> topicConfigurationException = Optional.empty();
+
+ ConfiguredTopology configuredTopology1 = new ConfiguredTopology(
+ topologyEpoch, subtopologies, internalTopicsToBeCreated, topicConfigurationException);
+ ConfiguredTopology configuredTopology2 = new ConfiguredTopology(
+ topologyEpoch, subtopologies, internalTopicsToBeCreated, topicConfigurationException);
+
+ assertEquals(configuredTopology1, configuredTopology2);
+ assertEquals(configuredTopology1.hashCode(), configuredTopology2.hashCode());
Review Comment:
Could you please add negative examples?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18311: Add internal datastructure for configuring topologies (1/N) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18268:
URL: https://github.com/apache/kafka/pull/18268#discussion_r1901004991
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/TopicConfigurationException.java:
##########
@@ -0,0 +1,73 @@
+/*
+ * 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;
+import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse.Status;
+
+import java.util.Objects;
+
+public class TopicConfigurationException extends RuntimeException {
+
+ private final Status status;
+
+ public TopicConfigurationException(StreamsGroupHeartbeatResponse.Status status, String message) {
+ super(message);
+ this.status = status;
+ }
+
+ public Status status() {
+ return status;
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final TopicConfigurationException that = (TopicConfigurationException) o;
+ return status == that.status && Objects.equals(getMessage(), that.getMessage());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(status, getMessage());
+ }
+
+ @Override
+ public String toString() {
+ return "TopicConfigurationException{" +
+ "status=" + status +
+ ", getMessage='" + getMessage() + '\'' +
+ '}';
+ }
Review Comment:
removed
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18311: Add internal datastructure for configuring topologies (1/N) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18268:
URL: https://github.com/apache/kafka/pull/18268#discussion_r1893965895
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredInternalTopicTest.java:
##########
@@ -0,0 +1,125 @@
+/*
+ * 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.InvalidTopicException;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ConfiguredInternalTopicTest {
+
+ @Test
+ public void testConstructorAndGetters() {
+ Map<String, String> topicConfigs = new HashMap<>();
+ topicConfigs.put("retention.ms", "1000");
+
+ ConfiguredInternalTopic config = new ConfiguredInternalTopic("test-topic", topicConfigs, Optional.of(3), Optional.of((short) 2));
+
+ assertEquals("test-topic", config.name());
+ assertEquals(topicConfigs, config.topicConfigs());
+ assertEquals(Optional.of(3), config.numberOfPartitions());
+ assertEquals(Optional.of((short) 2), config.replicationFactor());
+ }
Review Comment:
Let's keep it, now that we have it. It's logic that can be tested, and returning incorrect values from the getter does happen.
Once we make these classes immutable, we can move to record classes, where this kind of error does not happen anymore.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18311: Add internal datastructure for configuring topologies (1/N) [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #18268:
URL: https://github.com/apache/kafka/pull/18268#discussion_r1903887153
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredInternalTopic.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.internals.Topic;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * ConfiguredInternalTopic captures the properties required for configuring the internal topics we create for change-logs and repartitioning
+ * etc.
+ * <p>
+ * It is derived from the topology sent by the client, and the current state of the topics inside the broker. If the topics on the broker
+ * changes, the internal topic may need to be reconfigured.
+ */
+public record ConfiguredInternalTopic(String name,
+ Map<String, String> topicConfigs,
+ int numberOfPartitions,
+ Optional<Short> replicationFactor) {
Review Comment:
nit:
IMO the following order of parameters would be better:
```java
public record ConfiguredInternalTopic(String name,
int numberOfPartitions,
Optional<Short> replicationFactor,
Map<String, String> topicConfigs) {
```
Basically, first the mandatory parameters, then an optional specific configs, and then the rest.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredSubtopology.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.message.StreamsGroupDescribeResponseData;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Internal representation of a subtopology.
+ * <p>
+ * The subtopology is configured according to the number of partitions available in the source topics. It has regular expressions already
+ * resolved and defined exactly the information that is being used by streams groups assignment reconciliation.
+ * <p>
+ * Configured subtopologies may be recreated every time the input topics used by the subtopology are modified.
+ */
+public record ConfiguredSubtopology(Set<String> repartitionSinkTopics,
+ Set<String> sourceTopics,
+ Map<String, ConfiguredInternalTopic> repartitionSourceTopics,
+ Map<String, ConfiguredInternalTopic> stateChangelogTopics) {
Review Comment:
nit:
Also here I would prefer a different order. First source, then sink, finally changelog.
```suggestion
public record ConfiguredSubtopology(Set<String> sourceTopics,
Map<String, ConfiguredInternalTopic> repartitionSourceTopics,
Set<String> repartitionSinkTopics,
Map<String, ConfiguredInternalTopic> stateChangelogTopics) {
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredTopologyTest.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.message.CreateTopicsRequestData.CreatableTopic;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ConfiguredTopologyTest {
+
+ @Test
+ public void testIsReady() {
+ ConfiguredTopology readyTopology = new ConfiguredTopology(
+ 1, new HashMap<>(), new HashMap<>(), Optional.empty());
+ assertTrue(readyTopology.isReady());
+
+ ConfiguredTopology notReadyTopology = new ConfiguredTopology(
+ 1, new HashMap<>(), new HashMap<>(), Optional.of(TopicConfigurationException.missingSourceTopics("missing")));
+ assertFalse(notReadyTopology.isReady());
+ }
+
+ @Test
+ public void testAsStreamsGroupDescribeTopology() {
+ int topologyEpoch = 1;
Review Comment:
nit:
See my comment above about structure.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredInternalTopic.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.internals.Topic;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * ConfiguredInternalTopic captures the properties required for configuring the internal topics we create for change-logs and repartitioning
+ * etc.
+ * <p>
+ * It is derived from the topology sent by the client, and the current state of the topics inside the broker. If the topics on the broker
+ * changes, the internal topic may need to be reconfigured.
+ */
Review Comment:
Should we try to make the javadocs a bit more consistent?
For example:
```suggestion
* Captures the properties required for configuring the internal topics we create for changelogs and repartitioning etc.
* <p>
* It is derived from the topology sent by the client, and the current state of the topics inside the broker. If the topics on the broker
* changes, the internal topic may need to be reconfigured.
*/
```
To make it more similar to the javadocs of `ConfiguredSubtopology`.
Also, adding the docs for the parameters as you did in `ConfiguredTopology` makes sense.
A good class documentation is beneficial.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredSubtopologyTest.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * 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.message.StreamsGroupDescribeResponseData;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ConfiguredSubtopologyTest {
+
+ @Test
+ public void testAsStreamsGroupDescribeSubtopology() {
+ String subtopologyId = "subtopology1";
+
+ Set<String> sourceTopics = new HashSet<>(Set.of("sourceTopic1", "sourceTopic2"));
+ Set<String> repartitionSinkTopics = new HashSet<>(Set.of("repartitionSinkTopic1", "repartitionSinkTopic2"));
+
+ ConfiguredInternalTopic internalTopicMock = mock(ConfiguredInternalTopic.class);
+ StreamsGroupDescribeResponseData.TopicInfo topicInfo = new StreamsGroupDescribeResponseData.TopicInfo();
+ when(internalTopicMock.asStreamsGroupDescribeTopicInfo()).thenReturn(topicInfo);
+
+ Map<String, ConfiguredInternalTopic> repartitionSourceTopics = Map.of("repartitionSourceTopic1", internalTopicMock);
+ Map<String, ConfiguredInternalTopic> stateChangelogTopics = Map.of("stateChangelogTopic1", internalTopicMock);
+
+ ConfiguredSubtopology configuredSubtopology = new ConfiguredSubtopology(
+ repartitionSinkTopics, sourceTopics, repartitionSourceTopics, stateChangelogTopics);
Review Comment:
nit:
IMO, test code is more easily readable, if it has three blocks: first setup, then call under test, and finally verification. There are exception to this structure, but I believe this test would benefit from it.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredInternalTopic.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.internals.Topic;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * ConfiguredInternalTopic captures the properties required for configuring the internal topics we create for change-logs and repartitioning
Review Comment:
```suggestion
* ConfiguredInternalTopic captures the properties required for configuring the internal topics we create for changelogs and repartitioning
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18311: Add internal datastructure for configuring topologies (1/N) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18268:
URL: https://github.com/apache/kafka/pull/18268#discussion_r1903974852
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/topics/ConfiguredSubtopology.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.message.StreamsGroupDescribeResponseData;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Internal representation of a subtopology.
+ * <p>
+ * The subtopology is configured according to the number of partitions available in the source topics. It has regular expressions already
+ * resolved and defined exactly the information that is being used by streams groups assignment reconciliation.
+ * <p>
+ * Configured subtopologies may be recreated every time the input topics used by the subtopology are modified.
+ */
+public record ConfiguredSubtopology(Set<String> repartitionSinkTopics,
+ Set<String> sourceTopics,
Review Comment:
I added the null checks.
I think throwing on empty source topics is incorrect, since we don't require it in the KIP. So we'd need to update the KIP, otherwise the group coordinator will just crash on allowed topologies.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18311: Add internal datastructure for configuring topologies (1/N) [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #18268:
URL: https://github.com/apache/kafka/pull/18268
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org