This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 18337afa95b KAFKA-20430 Fix getTopicConfig() for connector-specific
offset stores (#22984)
18337afa95b is described below
commit 18337afa95b83893a330e38f5ad086f9b87d091b
Author: Kai Jay Tu <[email protected]>
AuthorDate: Wed Aug 5 18:38:58 2026 +0800
KAFKA-20430 Fix getTopicConfig() for connector-specific offset stores
(#22984)
Follow-up fix for #20384.
When `internal.topics.automatic.creation.enable` is set to `false` and a
connector-specific offset topic is missing, the error message
incorrectly references the worker-level config `offset.storage.topic`
instead of the connector-level config `offsets.storage.topic`. This is
because `readWriteStore()` and `readOnlyStore()` create anonymous
subclasses of `KafkaOffsetBackingStore` that do not override
`getTopicConfig()`.
This patch overrides `getTopicConfig()` in both factory methods to
return `SourceConnectorConfig.OFFSETS_TOPIC_CONFIG`
(`offsets.storage.topic`), so the error message correctly directs users
to the relevant connector configuration.
### Changes
- `KafkaOffsetBackingStore.java`: Override `getTopicConfig()` in
`readWriteStore()` and `readOnlyStore()` anonymous classes to return the
connector-level config key
- Added `KafkaOffsetBackingStoreGetTopicConfigTest.java` to verify the
fix
### Testing
Unit tests verify that:
1. Worker-level `KafkaOffsetBackingStore` returns `offset.storage.topic`
(unchanged behavior)
2. `readWriteStore()` now returns `offsets.storage.topic` (was
incorrectly returning `offset.storage.topic`)
3. `readOnlyStore()` now returns `offsets.storage.topic` (was
incorrectly returning `offset.storage.topic`)
Existing tests (`KafkaOffsetBackingStoreTest`,
`ConnectorOffsetBackingStoreTest`) pass with no regression.
Reviewers: Chia-Ping Tsai <[email protected]>
---
.../connect/storage/KafkaOffsetBackingStore.java | 11 +++
.../storage/KafkaTopicBasedBackingStore.java | 2 +-
.../KafkaOffsetBackingStoreGetTopicConfigTest.java | 83 ++++++++++++++++++++++
3 files changed, 95 insertions(+), 1 deletion(-)
diff --git
a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java
index 7920b3d6e0c..75de827f8c8 100644
---
a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java
+++
b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java
@@ -31,6 +31,7 @@ import
org.apache.kafka.common.serialization.ByteArrayDeserializer;
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 class KafkaOffsetBackingStore extends
KafkaTopicBasedBackingStore impleme
ignored -> true
);
}
+
+ @Override
+ protected String getTopicConfig() {
+ return SourceConnectorConfig.OFFSETS_TOPIC_CONFIG;
+ }
};
}
@@ -134,6 +140,11 @@ public class KafkaOffsetBackingStore extends
KafkaTopicBasedBackingStore impleme
ignored -> true
);
}
+
+ @Override
+ protected String getTopicConfig() {
+ return SourceConnectorConfig.OFFSETS_TOPIC_CONFIG;
+ }
};
}
diff --git
a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaTopicBasedBackingStore.java
b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaTopicBasedBackingStore.java
index f4215d95486..0cda2f46010 100644
---
a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaTopicBasedBackingStore.java
+++
b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaTopicBasedBackingStore.java
@@ -55,7 +55,7 @@ public abstract class KafkaTopicBasedBackingStore {
if (existing.isEmpty()) {
String msg = String.format("Topic '%s' specified via the
'%s' property is missing." +
" The config '%s' is set to '%s', so
automatic creation of internal topics is disabled." +
- " Either enable automatic creation or
create the topics manually before starting the worker.",
+ " Either enable automatic creation or
create the topics manually.",
topic, getTopicConfig(),
DistributedConfig.INTERNAL_TOPICS_AUTOMATIC_CREATION_ENABLE_CONFIG,
config.internalTopicsCreationEnabled());
throw new ConnectException(msg);
}
diff --git
a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreGetTopicConfigTest.java
b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreGetTopicConfigTest.java
new file mode 100644
index 00000000000..9ab6dba20f0
--- /dev/null
+++
b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreGetTopicConfigTest.java
@@ -0,0 +1,83 @@
+/*
+ * 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.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 testConnectorSpecificReadWriteStoreReturnsCorrectTopicConfig()
{
+ 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<>("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);
+
+ assertEquals(SourceConnectorConfig.OFFSETS_TOPIC_CONFIG,
connectorStore.getTopicConfig(),
+ "Connector-specific store should return
'offsets.storage.topic', not 'offset.storage.topic'");
+ }
+
+ @Test
+ public void testConnectorSpecificReadOnlyStoreReturnsCorrectTopicConfig() {
+ String connectorOffsetTopic = "my-connector-offsets";
+ Consumer<byte[], byte[]> consumer = new MockConsumer<>("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);
+
+ assertEquals(SourceConnectorConfig.OFFSETS_TOPIC_CONFIG,
connectorStore.getTopicConfig(),
+ "Connector-specific store should return
'offsets.storage.topic', not 'offset.storage.topic'");
+ }
+}
