Copilot commented on code in PR #22511: URL: https://github.com/apache/kafka/pull/22511#discussion_r3380139361
########## server/src/main/java/org/apache/kafka/server/streams/InMemoryTopologyDescriptionPlugin.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.server.streams; + +import org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription; +import org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin; + +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Reference {@link StreamsGroupTopologyDescriptionPlugin} that stores topology descriptions + * in an in-memory map. Intended for testing and as a starting point for real implementations; + * not suitable for production because its state is lost on broker restart and is not shared + * across brokers. + * + * <p>Stores at most one description per group, identified by its topology epoch. A + * {@code setTopology} call overwrites any previously stored description for that group; + * {@code getTopology} returns the stored description only when the requested topology epoch + * matches the stored epoch, and {@code null} otherwise. + */ +public class InMemoryTopologyDescriptionPlugin implements StreamsGroupTopologyDescriptionPlugin { + + private record Entry(int topologyEpoch, StreamsGroupTopologyDescription description) { } + + private final ConcurrentHashMap<String, Entry> store = new ConcurrentHashMap<>(); + + @Override + public void configure(Map<String, ?> configs) { + // No-op. + } + + @Override + public CompletableFuture<Void> setTopology(String groupId, int topologyEpoch, StreamsGroupTopologyDescription description) { + store.put(groupId, new Entry(topologyEpoch, description)); + return CompletableFuture.completedFuture(null); + } Review Comment: `StreamsGroupTopologyDescriptionPlugin` contract states implementations must not throw synchronously; however `store.put(groupId, ...)` will throw `NullPointerException` if `groupId` or `description` is null (ConcurrentHashMap disallows nulls). Consider validating inputs and returning a failed future instead so failures are always reported via exceptional completion. ########## server/src/main/java/org/apache/kafka/server/streams/InMemoryTopologyDescriptionPlugin.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.server.streams; + +import org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription; +import org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin; + +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Reference {@link StreamsGroupTopologyDescriptionPlugin} that stores topology descriptions + * in an in-memory map. Intended for testing and as a starting point for real implementations; + * not suitable for production because its state is lost on broker restart and is not shared + * across brokers. + * + * <p>Stores at most one description per group, identified by its topology epoch. A + * {@code setTopology} call overwrites any previously stored description for that group; + * {@code getTopology} returns the stored description only when the requested topology epoch + * matches the stored epoch, and {@code null} otherwise. + */ +public class InMemoryTopologyDescriptionPlugin implements StreamsGroupTopologyDescriptionPlugin { + + private record Entry(int topologyEpoch, StreamsGroupTopologyDescription description) { } + + private final ConcurrentHashMap<String, Entry> store = new ConcurrentHashMap<>(); + + @Override + public void configure(Map<String, ?> configs) { + // No-op. + } + + @Override + public CompletableFuture<Void> setTopology(String groupId, int topologyEpoch, StreamsGroupTopologyDescription description) { + store.put(groupId, new Entry(topologyEpoch, description)); + return CompletableFuture.completedFuture(null); + } + + @Override + public CompletableFuture<Void> deleteTopology(String groupId) { + store.remove(groupId); + return CompletableFuture.completedFuture(null); + } + + @Override + public CompletableFuture<StreamsGroupTopologyDescription> getTopology(String groupId, int topologyEpoch) { + Entry entry = store.get(groupId); + if (entry == null || entry.topologyEpoch() != topologyEpoch) { + return CompletableFuture.completedFuture(null); + } + return CompletableFuture.completedFuture(entry.description()); + } Review Comment: `getTopology` can throw synchronously if `groupId` is null (ConcurrentHashMap disallows null keys), but the SPI expects failures to be signaled via exceptional completion of the returned future. Return a failed future for invalid input instead of throwing. ########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java: ########## @@ -821,6 +826,90 @@ public void testDLQTopicNamePrefixCustomValue() { assertEquals("my-dlq-", config.errorsDLQTopicNamePrefix()); } + @Test + public void testStreamsGroupTopologyDescriptionPluginDefaultIsNull() { + GroupCoordinatorConfig config = createConfig(new HashMap<>()); + assertNull(config.streamsGroupTopologyDescriptionPlugin(Map.of())); + } Review Comment: This test asserts the default value is `null`, but the PR description indicates the config accessor should return `Optional<StreamsGroupTopologyDescriptionPlugin>` (implying `Optional.empty()` when unset). If the intended API is Optional, this test will need to be updated accordingly. ########## group-coordinator/group-coordinator-api/src/main/java/org/apache/kafka/coordinator/group/api/streams/StreamsGroupTopologyDescription.java: ########## @@ -0,0 +1,92 @@ +/* + * 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.api.streams; + +import java.util.Collection; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; + +/** + * Broker-side description of a Kafka Streams topology, as pushed by clients via + * {@code StreamsGroupTopologyDescriptionUpdate} and consumed by + * {@link StreamsGroupTopologyDescriptionPlugin} implementations. + * + * <p>This type mirrors {@code org.apache.kafka.streams.TopologyDescription} in shape + * but lives in {@code group-coordinator-api} so plugin implementations do not need + * to depend on {@code kafka-streams}. The wire schema only carries the successor + * relation; plugins that need both directions reconstruct predecessors in a single + * pass over the nodes. + */ +public record StreamsGroupTopologyDescription( Review Comment: PR description says the public Streams topology-description SPI is marked `@InterfaceStability.Evolving`, but `StreamsGroupTopologyDescription` itself is currently unannotated. Consider adding the annotation here (or at package level) to make the stability contract explicit for plugin implementers. ########## group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java: ########## @@ -897,6 +905,16 @@ protected List<ShareGroupPartitionAssignor> shareGroupAssignors( return assignors; } + public StreamsGroupTopologyDescriptionPlugin streamsGroupTopologyDescriptionPlugin( + Map<String, ?> additionalConfigs + ) { + Map<String, Object> overrides = new HashMap<>(additionalConfigs); + return config.getConfiguredInstance( + STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG, + StreamsGroupTopologyDescriptionPlugin.class, + overrides); + } Review Comment: PR description says the topology description plugin is instantiated and configured once and exposed as an `Optional<StreamsGroupTopologyDescriptionPlugin>`, but this accessor returns a nullable plugin and creates a new configured instance on every call via `AbstractConfig#getConfiguredInstance`. This can lead to multiple plugin instances (and potential resource leaks / inconsistent state) and contradicts the stated design. ########## server/src/main/java/org/apache/kafka/server/streams/InMemoryTopologyDescriptionPlugin.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.server.streams; + +import org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription; +import org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin; + +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Reference {@link StreamsGroupTopologyDescriptionPlugin} that stores topology descriptions + * in an in-memory map. Intended for testing and as a starting point for real implementations; + * not suitable for production because its state is lost on broker restart and is not shared + * across brokers. + * + * <p>Stores at most one description per group, identified by its topology epoch. A + * {@code setTopology} call overwrites any previously stored description for that group; + * {@code getTopology} returns the stored description only when the requested topology epoch + * matches the stored epoch, and {@code null} otherwise. + */ +public class InMemoryTopologyDescriptionPlugin implements StreamsGroupTopologyDescriptionPlugin { + + private record Entry(int topologyEpoch, StreamsGroupTopologyDescription description) { } + + private final ConcurrentHashMap<String, Entry> store = new ConcurrentHashMap<>(); + + @Override + public void configure(Map<String, ?> configs) { + // No-op. + } + + @Override + public CompletableFuture<Void> setTopology(String groupId, int topologyEpoch, StreamsGroupTopologyDescription description) { + store.put(groupId, new Entry(topologyEpoch, description)); + return CompletableFuture.completedFuture(null); + } + + @Override + public CompletableFuture<Void> deleteTopology(String groupId) { + store.remove(groupId); + return CompletableFuture.completedFuture(null); + } Review Comment: `deleteTopology` can throw synchronously if `groupId` is null (ConcurrentHashMap disallows null keys), which contradicts the SPI requirement to report failures via the returned future. Return a failed future for invalid input instead of throwing. ########## group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfigTest.java: ########## @@ -821,6 +826,90 @@ public void testDLQTopicNamePrefixCustomValue() { assertEquals("my-dlq-", config.errorsDLQTopicNamePrefix()); } + @Test + public void testStreamsGroupTopologyDescriptionPluginDefaultIsNull() { + GroupCoordinatorConfig config = createConfig(new HashMap<>()); + assertNull(config.streamsGroupTopologyDescriptionPlugin(Map.of())); + } + + @Test + public void testStreamsGroupTopologyDescriptionPluginLoadedAndConfigured() { + Map<String, Object> configs = new HashMap<>(); + configs.put(GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG, + TestTopologyDescriptionPlugin.class.getName()); + GroupCoordinatorConfig config = createConfig(configs); + + StreamsGroupTopologyDescriptionPlugin plugin = + config.streamsGroupTopologyDescriptionPlugin(Map.of()); + assertInstanceOf(TestTopologyDescriptionPlugin.class, plugin); + assertNotNull(((TestTopologyDescriptionPlugin) plugin).configs); + } + + @Test + public void testStreamsGroupTopologyDescriptionPluginAcceptsClassObject() { + Map<String, Object> configs = new HashMap<>(); + configs.put(GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG, + TestTopologyDescriptionPlugin.class); + GroupCoordinatorConfig config = createConfig(configs); + + assertInstanceOf(TestTopologyDescriptionPlugin.class, + config.streamsGroupTopologyDescriptionPlugin(Map.of())); + } + + @Test + public void testStreamsGroupTopologyDescriptionPluginReceivesAdditionalConfigs() { + Map<String, Object> configs = new HashMap<>(); + configs.put(GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG, + TestTopologyDescriptionPlugin.class); + GroupCoordinatorConfig config = createConfig(configs); + + try (TestTopologyDescriptionPlugin plugin = (TestTopologyDescriptionPlugin) + config.streamsGroupTopologyDescriptionPlugin(Map.of("injected.handle", "value"))) { + assertEquals("value", plugin.configs.get("injected.handle")); + } + } + + @Test + public void testStreamsGroupTopologyDescriptionPluginReturnsFreshInstancePerCall() { + Map<String, Object> configs = new HashMap<>(); + configs.put(GroupCoordinatorConfig.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_PLUGIN_CLASS_CONFIG, + TestTopologyDescriptionPlugin.class); + GroupCoordinatorConfig config = createConfig(configs); + + StreamsGroupTopologyDescriptionPlugin first = + config.streamsGroupTopologyDescriptionPlugin(Map.of()); + StreamsGroupTopologyDescriptionPlugin second = + config.streamsGroupTopologyDescriptionPlugin(Map.of()); + assertNotSame(first, second); + } Review Comment: This test enforces that `streamsGroupTopologyDescriptionPlugin(...)` returns a fresh instance on every call, which conflicts with the PR description stating the plugin is instantiated/configured once. If the plugin is meant to be a single broker-scoped instance, this expectation should change to assert the same instance is reused (and that it is closed on shutdown). -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
