Page History

package org.apache.kafka.connect.components;  
  
import org.apache.kafka.common.config.ConfigDef;  
  
/**  
* Interface for components that provide version and configuration specifications. 
* This interface establishes a common contract for all Kafka Connect components 
* that define a versin and expose configurable properties, enabling uniform discovery and introspection 
* of component configurations. 
* 
* <p>Components implementing this interface declare their version and configuration requirements 
* through a {@link ConfigDef} object, which describes the configuration properties 
* including their names, types, default values, validators, and documentation. 
  
* 
*/

public interface ConnectorPluginConnectPlugin {  

    /**
    * Get the version of this component.
    *
    * @return the version, formatted as a String. The version may not be {@code null} or empty.
    */
    String version();
  
	/**     
	* Returns the configuration specification for this component.     
	*     
	* <p>The returned {@link ConfigDef} object describes all configuration properties     
	* that this component accepts, including their types, default values, validators,     
	* importance levels, and documentation strings.     
	*     
	* @return the configuration definition for this component; never null     
	*/    
	ConfigDef config();
}  

public abstract class Connector implements ConnectorPluginConnectPlugin {  
  
    /**     
    * Define the configuration for the connector.     
    * @return The ConfigDef for this connector; may not be null.     
    */    
    @Override    
    public abstract ConfigDef config();
}  

public interface Converter extends Closeable, ConnectorPluginConnectPlugin {  
  
    /**     
    * Configuration specification for this converter.     
    * @return the configuration specification; may not be null     
    */    
    @Override    
    default ConfigDef config() { return new ConfigDef();}
}  

public interface HeaderConverter extends Configurable, Closeable, ConnectorPluginConnectPlugin {

    /**
     * Configuration specification for this set of header converters.
     * @return the configuration specification; may not be null
     */
	@Override
    ConfigDef config();
}

public interface Transformation<R extends ConnectRecord<R>> extends Configurable, Closeable, ConnectorPluginConnectPlugin {  
  
    /** Configuration specification for this transformation. 
    */    
    @Override    
    ConfigDef config();
}  

public interface Predicate<R extends ConnectRecord<R>> extends Configurable, AutoCloseable, ConnectorPluginConnectPlugin {  
  
    /**     
    * Configuration specification for this predicate.     
    *     
    * @return the configuration definition for this predicate; never null     
    */    
    @Override    
    ConfigDef config();
}  

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, ConnectorPluginConnectPlugin {


    /**
     * 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 policy override.     
    *     
    * @return the configuration definition for this policy override; never null     
    */    
    @Override    
    default ConfigDef config() { return new ConfigDef();}
}  

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, Closeable, ConnectorPluginConnectPlugin {

    /**
     * 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 rest extension.     
    *     
    * @return the configuration definition for this rest extension; never null     
    */    
    @Override    
    default ConfigDef config() { return new ConfigDef();}
}  

// Field with ConnectorPluginConnectPlugin type
private ConnectorPluginConnectPlugin specifier;

// Method parameter with ConnectorPluginConnectPlugin type
public MyConnector(ConnectorPlugin spec) { ... }

// Local variable with explicit ConnectorPluginConnectPlugin type
ConnectorPluginConnectPlugin spec = this;

// instanceof checks
if (obj instanceof ConnectorPluginConnectPlugin) { ... }

// Explicit cast to ConnectorPluginConnectPlugin
ConnectorPluginConnectPlugin casted = (ConnectorPluginConnectPlugin) obj;

// Explicitly implementing ConnectorPluginConnectPlugin
public class MyPlugin implements Transformation, ConnectorPluginConnectPlugin { ... }