KAFKA-20430 Fix getTopicConfig() for connector-specific offset stores

connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java

@@ -31,6 +31,7 @@
 import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.runtime.SourceConnectorConfig;
 import org.apache.kafka.connect.runtime.WorkerConfig;
 import org.apache.kafka.connect.runtime.distributed.DistributedConfig;
 import org.apache.kafka.connect.util.Callback;
@@ -101,6 +102,11 @@ public void configure(final WorkerConfig config) {
                         ignored -> true
                 );
             }
+
+            @Override
+            protected String getTopicConfig() {
+                return SourceConnectorConfig.OFFSETS_TOPIC_CONFIG;
+            }
         };
     }
 
@@ -134,6 +140,11 @@ public void configure(final WorkerConfig config) {
                         ignored -> true
                 );
             }
+
+            @Override
+            protected String getTopicConfig() {
+                return SourceConnectorConfig.OFFSETS_TOPIC_CONFIG;
+            }
         };
     }
 

connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreGetTopicConfigTest.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.connect.storage;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.MockConsumer;
+import org.apache.kafka.clients.consumer.OffsetResetStrategy;
+import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.connect.json.JsonConverter;
+import org.apache.kafka.connect.json.JsonConverterConfig;
+import org.apache.kafka.connect.runtime.SourceConnectorConfig;
+import org.apache.kafka.connect.runtime.distributed.DistributedConfig;
+import org.apache.kafka.connect.util.TopicAdmin;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+
+public class KafkaOffsetBackingStoreGetTopicConfigTest {
+
+    @Test
+    public void testWorkerLevelStoreReturnsCorrectTopicConfig() {
+        JsonConverter keyConverter = new JsonConverter();
+        keyConverter.configure(Collections.singletonMap(JsonConverterConfig.SCHEMAS_ENABLE_CONFIG, "false"), true);
+        KafkaOffsetBackingStore workerStore = new KafkaOffsetBackingStore(
+                () -> mock(TopicAdmin.class),
+                () -> "test-client-",
+                keyConverter
+        );
+        assertEquals(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG, workerStore.getTopicConfig(),
+                "Worker-level offset store should reference 'offset.storage.topic'");
+    }
+
+    @Test
+    public void testConnectorSpecificReadWriteStoreReturnsWrongTopicConfig() {
+        String connectorOffsetTopic = "my-connector-offsets";
+        Producer<byte[], byte[]> producer = new MockProducer<>(Cluster.empty(), false, null, new ByteArraySerializer(), new ByteArraySerializer());
+        Consumer<byte[], byte[]> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
+        TopicAdmin topicAdmin = mock(TopicAdmin.class);
+        JsonConverter keyConverter = new JsonConverter();
+        keyConverter.configure(Collections.singletonMap(JsonConverterConfig.SCHEMAS_ENABLE_CONFIG, "false"), true);
+
+        KafkaOffsetBackingStore connectorStore = KafkaOffsetBackingStore.readWriteStore(
+                connectorOffsetTopic, producer, consumer, topicAdmin, keyConverter);
+
+        String actual = connectorStore.getTopicConfig();
+        System.out.println("readWriteStore getTopicConfig() returns: '" + actual + "'");
+        System.out.println("Expected: '" + SourceConnectorConfig.OFFSETS_TOPIC_CONFIG + "'");
+
+        assertEquals(SourceConnectorConfig.OFFSETS_TOPIC_CONFIG, actual,
+                "Connector-specific store should return 'offsets.storage.topic', not 'offset.storage.topic'");
+    }
+
+    @Test
+    public void testConnectorSpecificReadOnlyStoreReturnsWrongTopicConfig() {
+        String connectorOffsetTopic = "my-connector-offsets";
+        Consumer<byte[], byte[]> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
+        TopicAdmin topicAdmin = mock(TopicAdmin.class);
+        JsonConverter keyConverter = new JsonConverter();
+        keyConverter.configure(Collections.singletonMap(JsonConverterConfig.SCHEMAS_ENABLE_CONFIG, "false"), true);
+
+        KafkaOffsetBackingStore connectorStore = KafkaOffsetBackingStore.readOnlyStore(
+                connectorOffsetTopic, consumer, topicAdmin, keyConverter);
+
+        String actual = connectorStore.getTopicConfig();
+        System.out.println("readOnlyStore getTopicConfig() returns: '" + actual + "'");
+        System.out.println("Expected: '" + SourceConnectorConfig.OFFSETS_TOPIC_CONFIG + "'");
+
+        assertEquals(SourceConnectorConfig.OFFSETS_TOPIC_CONFIG, actual,
+                "Connector-specific store should return 'offsets.storage.topic', not 'offset.storage.topic'");
+    }
+}

docs/getting-started/upgrade.md

@@ -49,7 +49,7 @@ type: docs
   * Brokers can now record a human-readable description of each streams group's processing topology via a pluggable backend, retrievable through `Admin#describeStreamsGroups` and `kafka-streams-groups.sh --describe --topology`. The feature is disabled unless the new broker configuration `group.streams.topology.description.plugin.class` is set to a `StreamsGroupTopologyDescriptionPlugin` implementation; on the client side, the new Kafka Streams configuration `topology.description.push.enabled` (default `true`) controls whether the client pushes topology descriptions when requested. This adds a new RPC, `StreamsGroupTopologyDescriptionUpdate`, bumps `StreamsGroupDescribe` and `StreamsGroupHeartbeat` to version 1, and introduces the error codes `GROUP_DELETION_FAILED` (134) and `STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED` (135). `DeleteGroups` is bumped to version 3, adding a per-group `ErrorMessage` field so brokers can report why a group deletion failed (for example, when the plugin fails to delete its stored topology description, the group is not deleted and `GROUP_DELETION_FAILED` is returned; retrying the deletion is safe). For further details, please refer to [KIP-1331](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1331%3A+Streams+Group+Topology+Description+Plugin) and the [Topology Description Plugin](/{version}/streams/developer-guide/topology-description-plugin/) documentation.
   * The `kafka-producer-perf-test.sh` tool now supports `--record-key-range`, `--key-distribution`, and `--random-seed` options to control the distribution of record keys. Use `--key-distribution range` for sequential key assignment (round-robin over the key range) or `--key-distribution random` for random key selection. The `--random-seed` option allows reproducible benchmark runs when using random key distribution. For further details, please refer to [KIP-1299](https://cwiki.apache.org/confluence/x/XpQ8G).
   * Share groups now support dead-letter queue functionality as outlined in [KIP-1191](https://cwiki.apache.org/confluence/x/fApJFg). Any records which are released (beyond max delivery count) or rejected by the share consumer become eligible for DLQ. Share group DLQ gets enabled when the Kafka feature `share.version` is upgraded to 2. The user can configure a DLQ topic on a share group by setting the dynamic config `errors.deadletterqueue.topic.name` (default `""`) to the name of the DLQ topic. The cluster can be configured to auto create the DLQ topics by setting the dynamic cluster config `errors.deadletterqueue.auto.create.topics.enable` to `true` (default `false`). If auto create is not enabled, the user must create the DLQ topic like a standard Kafka topic and set the dynamic config `errors.deadletterqueue.group.enable` to `true` on the DLQ topic. The DLQ topic name must be prefixed by the value set in the dynamic cluster config `errors.deadletterqueue.topic.name.prefix` (default `dlq.`). The records sent to the DLQ topic by default only contain source record metadata like group, topic name, partition id, offset and delivery count. If original record data is also required, the user must set the dynamic config `errors.deadletterqueue.copy.record.enable` to `true` on the share group.
-  * Kafka Connect distributed workers now support the `internal.topics.automatic.creation.enable` configuration (default: `true`). When set to `false`, Connect will not automatically create internal topics (offset, config, status, and connector-specific offset topics) and will instead fail at startup if any of these topics are missing. A new `connect-internal-topics.sh` tool is also available for manually creating these topics. For further details, please refer to [KIP-1209](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1209:+Add+configuration+to+control+internal+topic+creation+in+Kafka+Connect).
+  * Kafka Connect distributed workers now support the `internal.topics.automatic.creation.enable` configuration (default: `true`). When set to `false`, Connect will not automatically create internal topics (offset, config, status, and connector-specific offset topics) and will instead fail at startup if any of these topics are missing. A new `connect-internal-topics.sh` tool is also available for manually creating these topics. For further details, please refer to [KIP-1209](https://cwiki.apache.org/confluence/x/GAq2Fg).
 
 ## Upgrading to 4.3.0