This is an automated email from the ASF dual-hosted git repository.
zabetak pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/calcite.git
The following commit(s) were added to refs/heads/main by this push:
new a998e2743b [CALCITE-7770] Improve checks on KafkaTableFactory operands
a998e2743b is described below
commit a998e2743b80e6c502af1142326c4add10b4507e
Author: Stamatis Zampetakis <[email protected]>
AuthorDate: Tue Aug 18 19:11:19 2026 +0300
[CALCITE-7770] Improve checks on KafkaTableFactory operands
---
.../calcite/adapter/kafka/KafkaTableFactory.java | 25 +++-----
.../adapter/kafka/InvalidKafkaConsumer.java | 30 ++++++++++
.../calcite/adapter/kafka/InvalidKafkaOperand.java | 28 +++++++++
.../adapter/kafka/KafkaRowConverterTest.java | 2 +-
.../adapter/kafka/KafkaTableFactoryTest.java | 69 ++++++++++++++++++++++
site/_docs/history.md | 4 ++
6 files changed, 139 insertions(+), 19 deletions(-)
diff --git
a/kafka/src/main/java/org/apache/calcite/adapter/kafka/KafkaTableFactory.java
b/kafka/src/main/java/org/apache/calcite/adapter/kafka/KafkaTableFactory.java
index f6cbce5353..2d6c3a2a10 100644
---
a/kafka/src/main/java/org/apache/calcite/adapter/kafka/KafkaTableFactory.java
+++
b/kafka/src/main/java/org/apache/calcite/adapter/kafka/KafkaTableFactory.java
@@ -16,6 +16,7 @@
*/
package org.apache.calcite.adapter.kafka;
+import org.apache.calcite.avatica.AvaticaUtils;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.calcite.schema.TableFactory;
@@ -51,20 +52,7 @@ public KafkaTableFactory() {
final KafkaRowConverter rowConverter;
if (operand.containsKey(KafkaTableConstants.SCHEMA_ROW_CONVERTER)) {
String rowConverterClass = (String)
operand.get(KafkaTableConstants.SCHEMA_ROW_CONVERTER);
- try {
- final Class<?> klass = Class.forName(rowConverterClass);
- rowConverter = (KafkaRowConverter)
klass.getDeclaredConstructor().newInstance();
- } catch (InstantiationException | InvocationTargetException
- | IllegalAccessException | ClassNotFoundException
- | NoSuchMethodException e) {
- final String details =
- String.format(Locale.ROOT,
- "Failed to create table '%s' with configuration:\n"
- + "'%s'\n"
- + "KafkaRowConverter '%s' is invalid",
- name, operand, rowConverterClass);
- throw new RuntimeException(details, e);
- }
+ rowConverter = AvaticaUtils.instantiatePlugin(KafkaRowConverter.class,
rowConverterClass);
} else {
rowConverter = new KafkaRowConverterImpl();
}
@@ -77,12 +65,13 @@ public KafkaTableFactory() {
if (operand.containsKey(KafkaTableConstants.SCHEMA_CUST_CONSUMER)) {
String custConsumerClass = (String)
operand.get(KafkaTableConstants.SCHEMA_CUST_CONSUMER);
try {
+ Class<? extends Consumer> klass =
+ Class.forName(custConsumerClass, false,
KafkaTableFactory.class.getClassLoader())
+ .asSubclass(Consumer.class);
tableOptionBuilder.setConsumer(
- (Consumer) Class.forName(custConsumerClass)
- .getConstructor(OffsetResetStrategy.class)
- .newInstance(OffsetResetStrategy.NONE));
+
klass.getConstructor(OffsetResetStrategy.class).newInstance(OffsetResetStrategy.NONE));
} catch (ClassNotFoundException | NoSuchMethodException |
IllegalAccessException
- | InstantiationException | InvocationTargetException e) {
+ | InstantiationException | InvocationTargetException |
ClassCastException e) {
final String details =
String.format(Locale.ROOT,
"Fail to create table '%s' with configuration:\n"
diff --git
a/kafka/src/test/java/org/apache/calcite/adapter/kafka/InvalidKafkaConsumer.java
b/kafka/src/test/java/org/apache/calcite/adapter/kafka/InvalidKafkaConsumer.java
new file mode 100644
index 0000000000..58612d2ac9
--- /dev/null
+++
b/kafka/src/test/java/org/apache/calcite/adapter/kafka/InvalidKafkaConsumer.java
@@ -0,0 +1,30 @@
+/*
+ * 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.calcite.adapter.kafka;
+
+import org.apache.kafka.clients.consumer.OffsetResetStrategy;
+
+/**
+ * An invalid class that always fails if instantiated.
+ */
+@SuppressWarnings("unused")
+public final class InvalidKafkaConsumer {
+
+ public InvalidKafkaConsumer(OffsetResetStrategy strategy) {
+ throw new AssertionError("Class must not be instantiated.");
+ }
+}
diff --git
a/kafka/src/test/java/org/apache/calcite/adapter/kafka/InvalidKafkaOperand.java
b/kafka/src/test/java/org/apache/calcite/adapter/kafka/InvalidKafkaOperand.java
new file mode 100644
index 0000000000..42af807984
--- /dev/null
+++
b/kafka/src/test/java/org/apache/calcite/adapter/kafka/InvalidKafkaOperand.java
@@ -0,0 +1,28 @@
+/*
+ * 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.calcite.adapter.kafka;
+
+/**
+ * An invalid class that always fail if instantiated.
+ */
+@SuppressWarnings("unused")
+public final class InvalidKafkaOperand {
+
+ public InvalidKafkaOperand() {
+ throw new AssertionError("Class must not be instantiated.");
+ }
+}
diff --git
a/kafka/src/test/java/org/apache/calcite/adapter/kafka/KafkaRowConverterTest.java
b/kafka/src/test/java/org/apache/calcite/adapter/kafka/KafkaRowConverterTest.java
index ab2ecbfe28..7b9e73caa1 100644
---
a/kafka/src/test/java/org/apache/calcite/adapter/kafka/KafkaRowConverterTest.java
+++
b/kafka/src/test/java/org/apache/calcite/adapter/kafka/KafkaRowConverterTest.java
@@ -28,7 +28,7 @@
* Implementation of {@link KafkaRowConverter} for testing. Both key and value
* are saved as {@code byte[]}.
*/
-class KafkaRowConverterTest implements KafkaRowConverter<String, String> {
+public class KafkaRowConverterTest implements KafkaRowConverter<String,
String> {
/**
* Generates a row schema for a given Kafka topic.
*
diff --git
a/kafka/src/test/java/org/apache/calcite/adapter/kafka/KafkaTableFactoryTest.java
b/kafka/src/test/java/org/apache/calcite/adapter/kafka/KafkaTableFactoryTest.java
new file mode 100644
index 0000000000..d4511f0208
--- /dev/null
+++
b/kafka/src/test/java/org/apache/calcite/adapter/kafka/KafkaTableFactoryTest.java
@@ -0,0 +1,69 @@
+/*
+ * 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.calcite.adapter.kafka;
+
+import org.apache.calcite.jdbc.CalciteSchema;
+import org.apache.calcite.schema.SchemaPlus;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for {@link KafkaTableFactory}.
+ */
+public class KafkaTableFactoryTest {
+ /**
+ * Test case for
+ * <a href="https://issues.apache.org/jira/browse/CALCITE-7770";>
+ * Improve checks on KafkaTableFactory operands</a>. The test verifies
+ * if specified converter class does not implement the appropriate interface
+ * the method throws the appropriate exception. */
+ @Test void testKafkaTableFactoryCreateInvalidRowConverterThrows() {
+ KafkaTableFactory factory = new KafkaTableFactory();
+ SchemaPlus schema = CalciteSchema.createRootSchema(false).plus();
+ String invalidConverter =
"org.apache.calcite.adapter.kafka.InvalidKafkaOperand";
+ Map<String, Object> operand =
+ Collections.singletonMap(KafkaTableConstants.SCHEMA_ROW_CONVERTER,
invalidConverter);
+ RuntimeException ex =
+ assertThrows(RuntimeException.class, () -> factory.create(schema, "t",
operand, null));
+ assertTrue(ex.getMessage()
+ .contains("not valid for plugin type
org.apache.calcite.adapter.kafka.KafkaRowConverter"));
+ }
+
+ /**
+ * Test case for
+ * <a href="https://issues.apache.org/jira/browse/CALCITE-7770";>
+ * Improve checks on KafkaTableFactory operands</a>. The test verifies
+ * if specified consumer class does not implement the appropriate interface
+ * the method throws the appropriate exception. */
+ @Test void testKafkaTableFactoryCreateInvalidConsumerThrows() {
+ KafkaTableFactory factory = new KafkaTableFactory();
+ SchemaPlus schema = CalciteSchema.createRootSchema(false).plus();
+ String invalidConsumer =
"org.apache.calcite.adapter.kafka.InvalidKafkaConsumer";
+ Map<String, Object> operand =
+ Collections.singletonMap(KafkaTableConstants.SCHEMA_CUST_CONSUMER,
invalidConsumer);
+ RuntimeException ex =
+ assertThrows(RuntimeException.class, () -> factory.create(schema, "t",
operand, null));
+ assertInstanceOf(ClassCastException.class, ex.getCause());
+ }
+}
diff --git a/site/_docs/history.md b/site/_docs/history.md
index d1e0caae3a..9ba5c7a42a 100644
--- a/site/_docs/history.md
+++ b/site/_docs/history.md
@@ -59,6 +59,10 @@ #### Breaking Changes
`ReflectiveSchema.Input` annotation. If the annotation is not present the
creation will fail with `IllegalArgumentException: class X is not annotated
@Input`.
+* [<a
href="https://issues.apache.org/jira/browse/CALCITE-7770";>CALCITE-7770</a>]
+`KafkaTableFactory` row converter operand now requires the presence of a
`public`
+constructor for instantiating the class.
+
* [<a
href="https://issues.apache.org/jira/browse/CALCITE-7713";>CALCITE-7713</a>]
Class loading from model files has been disabled by default. Any attempt to
load
classes from model files will lead to `SecurityException` unless an appropriate
