DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
| Code Block | ||||
|---|---|---|---|---|
| ||||
public interface Predicate<R extends ConnectRecord<R>> extends Configurable, AutoCloseable, ConfigSpecifier {
/**
* Configuration specification for this predicate.
*
* @return the configuration definition for this predicate; never null
*/
@Override
ConfigDef config();
} |
Other interfaces like ConnectRestExtension and ConnectorClientConfigOverridePolicy, even if they currently don't declare a config() method, since they implements the Configurable we can think to add the ConfigSpecifier for future use in this way
| Code Block | ||||
|---|---|---|---|---|
| ||||
package org.apache.kafka.connect.connector.policy; import org.apache.kafka.common.Configurable; import org.apache.kafka.common.config.ConfigValue; import java.util.List; /** * An interface for enforcing a policy on overriding of Kafka client configs via the connector configs. * <p> * Common use cases are ability to provide principal per connector, <code>sasl.jaas.config</code> * and/or enforcing that the producer/consumer configurations for optimizations are within acceptable ranges. * <p>Kafka Connect discovers implementations of this interface using the Java {@link java.util.ServiceLoader} mechanism. * To support this, implementations of this interface should also contain a service provider configuration file in * {@code META-INF/services/org.apache.kafka.connect.connector.policy.ConnectorClientConfigOverridePolicy}. * <p> * Implement {@link org.apache.kafka.common.metrics.Monitorable} to enable the policy to register metrics. * The following tags are automatically added to all metrics registered: <code>config</code> set to * <code>connector.client.config.override.policy</code>, and <code>class</code> set to the * ConnectorClientConfigOverridePolicy class name. */ public interface ConnectorClientConfigOverridePolicy extends Configurable, AutoCloseable, ConfigSpecifier { /** * Workers will invoke this before configuring per-connector Kafka admin, producer, and consumer client instances * to validate if all the overridden client configurations are allowed per the policy implementation. * This would also be invoked during the validation of connector configs via the REST API. * <p> * If there are any policy violations, the connector will not be started. * * @param connectorClientConfigRequest an instance of {@link ConnectorClientConfigRequest} that provides the configs * to be overridden and its context; never {@code null} * @return list of {@link ConfigValue} instances that describe each client configuration in the request and includes an {@link ConfigValue#errorMessages() error} if the configuration is not allowed by the policy; never null */ List<ConfigValue> validate(ConnectorClientConfigRequest connectorClientConfigRequest); /** * Configuration specification for this predicate. * * @return the configuration definition for this predicate; never null */ @Override default ConfigDef config() { return new ConfigDef();} } |
| Code Block | ||||
|---|---|---|---|---|
| ||||
package org.apache.kafka.connect.rest; import org.apache.kafka.common.Configurable; import org.apache.kafka.connect.components.Versioned; import org.apache.kafka.connect.health.ConnectClusterState; import java.io.Closeable; import java.util.Map; /** * A plugin interface to allow registration of new JAX-RS resources like Filters, REST endpoints, providers, etc. The implementations will * be discovered using the standard Java {@link java.util.ServiceLoader} mechanism by Connect's plugin class loading mechanism. * * <p>Kafka Connect discovers implementations of this interface using the Java {@link java.util.ServiceLoader} mechanism. * To support this, implementations of this interface should also contain a service provider configuration file in * {@code META-INF/services/org.apache.kafka.connect.rest.ConnectRestExtension}. * <p>The extension class(es) must be packaged as a plugin, including the JARs of all dependencies except those * already provided by the Connect framework. * * <p>To install into a Connect installation, add a directory named for the plugin and containing the plugin's JARs into a directory that is * on Connect's {@code plugin.path}, and (re)start the Connect worker. * * <p>When the Connect worker process starts up, it will read its configuration and instantiate all of the REST extension implementation * classes that are specified in the `rest.extension.classes` configuration property. Connect will then pass its configuration to each * extension via the {@link Configurable#configure(Map)} method, and will then call {@link #register} with a provided context. * * <p>When the Connect worker shuts down, it will call the extension's {@link #close} method to allow the implementation to release all of * its resources. * * <p>Implement {@link org.apache.kafka.common.metrics.Monitorable} to enable the extension to register metrics. * The following tags are automatically added to all metrics registered: <code>config</code> set to * <code>rest.extension.classes</code>, and <code>class</code> set to the ConnectRestExtension class name. */ public interface ConnectRestExtension extends Configurable, Versioned, Closeable, ConfigSpecifier { /** * ConnectRestExtension implementations can register custom JAX-RS resources via this method. The Connect framework * will invoke this method after registering the default Connect resources. If the implementations attempt * to re-register any of the Connect resources, it will be ignored and will be logged. * * @param restPluginContext The context provides access to JAX-RS {@link jakarta.ws.rs.core.Configurable} and {@link * ConnectClusterState}.The custom JAX-RS resources can be registered via the {@link * ConnectRestExtensionContext#configurable()} */ void register(ConnectRestExtensionContext restPluginContext); /** * Configuration specification for this predicate. * * @return the configuration definition for this predicate; never null */ @Override default ConfigDef config() { return new ConfigDef();} } |
Proposed Changes
Implementation Plan
1. Introduce ConfigSpecifier Interface
- Add the new org.apache.kafka.connect.components.ConfigSpecifier interface to the connect-api module
- Include a comprehensive JavaDoc explaining the purpose and usage
2. Update Existing Interfaces
- Modify Connector, Converter, Transformation, and Predicate, ConnectorClientConfigOverridePolicy, and ConnectRestExtension to extend ConfigSpecifier
- Preserve all existing method signatures and semantics
...
No deprecation is necessary. The existing config() methods methods in individual interfaces remain valid; they simply gain a common ancestor.
...