yashmayya commented on code in PR #14304:
URL: https://github.com/apache/kafka/pull/14304#discussion_r1309974448
##########
clients/src/main/java/org/apache/kafka/common/utils/Utils.java:
##########
@@ -1090,6 +1090,23 @@ public interface UncheckedCloseable extends
AutoCloseable {
void close();
}
+ /**
+ * Closes {@code maybeCloseable} if it implements the {@link
AutoCloseable} interface,
+ * and if an exception is thrown, it is logged at the WARN level.
+ * <b>Be cautious when passing method references as an argument.</b> For
example:
+ * <p>
+ * {@code closeQuietly(task::stop, "source task");}
+ * <p>
+ * Although this method gracefully handles null {@link AutoCloseable}
objects, attempts to take a method
Review Comment:
```suggestion
* Although this method gracefully handles null objects, attempts to
take a method
```
nit
##########
connect/runtime/src/main/java/org/apache/kafka/connect/util/ConcreteSubClassValidator.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.connect.util;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.utils.Utils;
+
+import java.lang.reflect.Modifier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class ConcreteSubClassValidator implements ConfigDef.Validator {
+ private final Class<?> expectedSuperClass;
+
+ private ConcreteSubClassValidator(Class<?> expectedSuperClass) {
+ this.expectedSuperClass = expectedSuperClass;
+ }
+
+ public static ConcreteSubClassValidator forSuperClass(Class<?>
expectedSuperClass) {
+ return new ConcreteSubClassValidator(expectedSuperClass);
+ }
+
+ @Override
+ public void ensureValid(String name, Object value) {
+ if (value == null) {
+ // The value will be null if the class couldn't be found; no point
in performing follow-up validation
+ return;
+ }
+
+ Class<?> cls = (Class<?>) value;
+ if (!expectedSuperClass.isAssignableFrom(cls)) {
+ throw new ConfigException(name, String.valueOf(cls), "Not a " +
expectedSuperClass.getSimpleName());
+ }
+
+ if (Modifier.isAbstract(cls.getModifiers())) {
Review Comment:
Could this be extracted out into a common util method and re-used across
here and
https://github.com/apache/kafka/blob/dbda60c60da8f5a7eabe113615196b729b40a0e8/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorConfig.java#L497-L508?
##########
connect/runtime/src/main/java/org/apache/kafka/connect/util/InstantiableClassValidator.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * 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.util;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.utils.Utils;
+
+public class InstantiableClassValidator implements ConfigDef.Validator {
+
+ @Override
+ public void ensureValid(String name, Object value) {
+ if (value == null) {
+ // The value will be null if the class couldn't be found; no point
in performing follow-up validation
+ return;
+ }
+
+ Class<?> cls = (Class<?>) value;
+ try {
+ Object o = cls.getDeclaredConstructor().newInstance();
+ Utils.maybeCloseQuietly(o, o + " (instantiated for preflight
validation");
+ } catch (NoSuchMethodException e) {
+ throw new ConfigException(name, cls.getName(), "Could not find a
public no-argument constructor for class" + (e.getMessage() != null ? ": " +
e.getMessage() : ""));
+ } catch (ReflectiveOperationException | RuntimeException e) {
+ throw new ConfigException(name, cls.getName(), "Could not
instantiate class" + (e.getMessage() != null ? ": " + e.getMessage() : ""));
+ }
Review Comment:
Could we leverage
[Utils::newInstance](https://github.com/apache/kafka/blob/dbda60c60da8f5a7eabe113615196b729b40a0e8/clients/src/main/java/org/apache/kafka/common/utils/Utils.java#L390-L403)
here?
##########
connect/runtime/src/main/java/org/apache/kafka/connect/util/ConcreteSubClassValidator.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.connect.util;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.utils.Utils;
+
+import java.lang.reflect.Modifier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class ConcreteSubClassValidator implements ConfigDef.Validator {
+ private final Class<?> expectedSuperClass;
+
+ private ConcreteSubClassValidator(Class<?> expectedSuperClass) {
+ this.expectedSuperClass = expectedSuperClass;
+ }
+
+ public static ConcreteSubClassValidator forSuperClass(Class<?>
expectedSuperClass) {
+ return new ConcreteSubClassValidator(expectedSuperClass);
+ }
+
+ @Override
+ public void ensureValid(String name, Object value) {
+ if (value == null) {
+ // The value will be null if the class couldn't be found; no point
in performing follow-up validation
+ return;
+ }
Review Comment:
Any idea why the validator is invoked even if the class can't be found for a
class type config? 🤔
Is this intentional or a bug?
##########
clients/src/main/java/org/apache/kafka/common/utils/Utils.java:
##########
@@ -1090,6 +1090,23 @@ public interface UncheckedCloseable extends
AutoCloseable {
void close();
}
+ /**
+ * Closes {@code maybeCloseable} if it implements the {@link
AutoCloseable} interface,
+ * and if an exception is thrown, it is logged at the WARN level.
+ * <b>Be cautious when passing method references as an argument.</b> For
example:
+ * <p>
+ * {@code closeQuietly(task::stop, "source task");}
Review Comment:
```suggestion
* {@code maybeCloseQuietly(task::stop, "source task");}
```
##########
connect/runtime/src/main/java/org/apache/kafka/connect/util/ConcreteSubClassValidator.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.connect.util;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.utils.Utils;
+
+import java.lang.reflect.Modifier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class ConcreteSubClassValidator implements ConfigDef.Validator {
+ private final Class<?> expectedSuperClass;
+
+ private ConcreteSubClassValidator(Class<?> expectedSuperClass) {
+ this.expectedSuperClass = expectedSuperClass;
+ }
+
+ public static ConcreteSubClassValidator forSuperClass(Class<?>
expectedSuperClass) {
+ return new ConcreteSubClassValidator(expectedSuperClass);
+ }
+
+ @Override
+ public void ensureValid(String name, Object value) {
+ if (value == null) {
+ // The value will be null if the class couldn't be found; no point
in performing follow-up validation
+ return;
+ }
Review Comment:
Any idea why the validator is invoked even if the class can't be found for a
class type config? 🤔
Is this intentional or a bug?
##########
connect/runtime/src/main/java/org/apache/kafka/connect/util/InstantiableClassValidator.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * 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.util;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.utils.Utils;
+
+public class InstantiableClassValidator implements ConfigDef.Validator {
+
+ @Override
+ public void ensureValid(String name, Object value) {
+ if (value == null) {
+ // The value will be null if the class couldn't be found; no point
in performing follow-up validation
+ return;
+ }
+
+ Class<?> cls = (Class<?>) value;
+ try {
+ Object o = cls.getDeclaredConstructor().newInstance();
+ Utils.maybeCloseQuietly(o, o + " (instantiated for preflight
validation");
+ } catch (NoSuchMethodException e) {
+ throw new ConfigException(name, cls.getName(), "Could not find a
public no-argument constructor for class" + (e.getMessage() != null ? ": " +
e.getMessage() : ""));
+ } catch (ReflectiveOperationException | RuntimeException e) {
+ throw new ConfigException(name, cls.getName(), "Could not
instantiate class" + (e.getMessage() != null ? ": " + e.getMessage() : ""));
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "A class with a public, no-argument constructor";
+ }
+}
Review Comment:
nit: missing newline at the end of the file
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorConfig.java:
##########
@@ -82,17 +86,29 @@ public class ConnectorConfig extends AbstractConfig {
public static final String KEY_CONVERTER_CLASS_CONFIG =
WorkerConfig.KEY_CONVERTER_CLASS_CONFIG;
public static final String KEY_CONVERTER_CLASS_DOC =
WorkerConfig.KEY_CONVERTER_CLASS_DOC;
public static final String KEY_CONVERTER_CLASS_DISPLAY = "Key converter
class";
+ private static final ConfigDef.Validator KEY_CONVERTER_CLASS_VALIDATOR =
ConfigDef.CompositeValidator.of(
+ ConcreteSubClassValidator.forSuperClass(Converter.class),
+ new InstantiableClassValidator()
+ );
public static final String VALUE_CONVERTER_CLASS_CONFIG =
WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG;
public static final String VALUE_CONVERTER_CLASS_DOC =
WorkerConfig.VALUE_CONVERTER_CLASS_DOC;
public static final String VALUE_CONVERTER_CLASS_DISPLAY = "Value
converter class";
+ private static final ConfigDef.Validator VALUE_CONVERTER_CLASS_VALIDATOR =
ConfigDef.CompositeValidator.of(
+ ConcreteSubClassValidator.forSuperClass(Converter.class),
+ new InstantiableClassValidator()
+ );
public static final String HEADER_CONVERTER_CLASS_CONFIG =
WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG;
public static final String HEADER_CONVERTER_CLASS_DOC =
WorkerConfig.HEADER_CONVERTER_CLASS_DOC;
public static final String HEADER_CONVERTER_CLASS_DISPLAY = "Header
converter class";
// The Connector config should not have a default for the header
converter, since the absence of a config property means that
// the worker config settings should be used. Thus, we set the default to
null here.
public static final String HEADER_CONVERTER_CLASS_DEFAULT = null;
+ private static final ConfigDef.Validator HEADER_CONVERTER_CLASS_VALIDATOR
= ConfigDef.CompositeValidator.of(
+ ConcreteSubClassValidator.forSuperClass(HeaderConverter.class),
+ new InstantiableClassValidator()
Review Comment:
Do we really need these to be separate validators if they're always used
together? Are there any use cases where it'd be beneficial for these to be in
separate validator classes?
##########
connect/runtime/src/main/java/org/apache/kafka/connect/util/ConcreteSubClassValidator.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.connect.util;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.utils.Utils;
+
+import java.lang.reflect.Modifier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class ConcreteSubClassValidator implements ConfigDef.Validator {
+ private final Class<?> expectedSuperClass;
+
+ private ConcreteSubClassValidator(Class<?> expectedSuperClass) {
+ this.expectedSuperClass = expectedSuperClass;
+ }
+
+ public static ConcreteSubClassValidator forSuperClass(Class<?>
expectedSuperClass) {
Review Comment:
I'm curious - why do we need this if there's only a single constructor
anyway (i.e. why not just use a public constructor instead)?
--
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]
