ConfigurationFragments: A strategy to share config properties across class hierarchies.

[Claudenw]

Problem Statement

We have a need to add configuration definitions to the system in general and in particular we need to be able to share configuration properties across the Source and Sink implementations. A problem arises because Kafka Connector framework defines two distinct base classes for Source (SourceConnectorConfig) and Sink (SinkConnectorConfig). These share a common Kafka Connector parent class (ConnectorConfig). These configuration use a ConfigDef object and modify it to create an internal representation of ConfigDef that is then used to answer any of the "getX()" methods (e.g. getInt(String key), get(String key), etc.). This means that the ConfigDef has to be constructed and inserted into the constructors for the SourceConnectorConfig and SinkConnectorConfig as any later changes to the ConfigDef will be ignored.

In pull request #312 a set of configuration classes has been proposed. Under this proposal, if we look at a typical Source connector, for example S3 Source Connector we find the following Config hierarchy: ConnectorConfig -> SourceConnectorConfig -> CommonSourceConfig -> S3SourceConfig; where CommonSourceConfig and S3SourceConfig is in our code base.

A typical Sink connector has a Config hierarchy like: ConnectorConfig -> SinkConnectorConfig -> CommonSinkConfig -> S3SinkConfig. Similarly CommonSinkConfig and S3SinkConfig are in our code base.

The problem arises for when S3SinkConfig and S3SourceConfig need to share common property definitions and access methods.

We need a mechanism that will:

1. Add configuration property definitions to the ConfigDef before the constructor is called.
2. Create standard methods that can be called from either the Source or Sink hierarchies, as part of the DRY principle.
3. Not break the current implementations.

Proposed Solution

Create Configuration Fragments

The concept is that the shared configuration properties can be broken down in to logical fragments that handle single concept. Those fragments will provide the Configuration Properties and standard methods to extract data from the configuration.

This implementation must able to fit into#312 to create a shared connector configuration framework for the connectors in this repository.

As an example let's look at the SinkCommonsConfig from #312 which implements a hierarchy to be used across the connectors without requiring modification of all existing connectors while providing an easy upgrade path for the code.

ConfigFragment Class

This class provides the basis of the framework for the shared methods.

public class ConfigFragment {

    protected final AbstractConfig cfg;

    protected ConfigFragment(final AbstractConfig cfg) {
        this.cfg = cfg;
    }
}

CommonConfigFragment Class

This is an example of an implementation of ConfigFragment that is used in the SinkCommonsConfig

public final class CommonConfigFragment extends ConfigFragment {
 
    // private static String used in configuraiton.
    private static final String GROUP_COMPRESSION = "File Compression";
   ...

    public CommonConfigFragment(AbstractConfig cfg) {
        super(cfg);
    }

    // A method to add the this fragments configuration properties to the ConfigDef.
    static void update(ConfigDef configDef) {
        configDef.define(KAFKA_RETRY_BACKOFF_MS_CONFIG, ConfigDef.Type.LONG, null, new ConfigDef.Validator() {
        ...
    }

    // The method that this configuration supplies.  Fragments may provide multiple methods.
    public Long getKafkaRetryBackoffMs() {
        return cfg.getLong(KAFKA_RETRY_BACKOFF_MS_CONFIG);
    }
}

SinkCommonConfig

This is a modified version of the code from #312. Not all code is present in this example, it is for discussion illumination only. The SourceCommonsConfig would be very similar.

// extends the Kafka SinkConnectorConfig class.
public class SinkCommonConfig extends SinkConnectorConfig  {

    private final CommonConfigFragment commonConfig;

    private static Plugins update(ConfigDef configDef, Map<String, String> props) {
        CommonConfigFragment.update(configDef);
        // FIXME this is a hack to get the configuration updated in the SinkConnectorConfig
        configDef.configKeys().values().forEach(def -> SinkCommonConfig.configDef().define(def));
        return new Plugins(props);
    }

    public SinkCommonConfig(final ConfigDef definition, final Map<String, String> props) { 
        super(update(definition, props), props); // <- note call to hacky update method
        // construct the fragments that his config supports
        commonConfig = new CommonConfigFragment(this);
    }

    // delegate to CommonsConfigFragment method, use to support existing code.
    /**
     * @deprecated use getCommonsConfig().getgetKafkaRetryBackoffMs()
     */
     @Deprecated
    public Long getKafkaRetryBackoffMs() {
        return commonConfig.getKafkaRetryBackoffMs();
    }

    // method to retrieve CommonsConfig object for fluent use
    public final CommonsConfigFragment getCommonsConfig() {
        return commonConfig;
    }
}

S3SinkConfig

This is not based on existing code but is an example of a clean implementation of the additional functionality mix-in. It assumes the existence of S3AuthenticationFragment which has not been presented here. And assumes that the S3SinkConnector will utilise the fluent access methods.

public final class S3SinkConfig extends SinkCommonConfig {

    private final S3AuthenticationFragment s3Auth;

    private static ConfigDef createConfig() {
        ConfigDef configDef = new ConfigDef()
        S3AuthenticaitonFragment.update(configDef);
        return configDef;
   }
   
    S3SinkConfig(Map<String, String> props) {
        super(createConfig(), props);
        s3Auth = new S3AuthenticationFragment(this);
   }

    public final S3AuthenticationFragment getAuthFragment() {
        return s3Auth;
    }
}

Advantages of Config Fragments

• Meets the requirements specified above.
• Isolate the Static strings that are used in the configuration definition to a single class.
• Provide single canonical implementation of configuration property access that is used across multiple Configurations.
• Configuration changes driven by code changes are isolated and do not necessitate other changes across the hierarchy.
• Provides a path to remove existing code clutter.

Rejected Alternatives

1. Dynamic Proxies: This solution will work but dynamic proxies are hard to debug/step through and will add significant complexity to the solution.
2. Builder strategy: In many ways this solution would be improved by the use of a ConfigDef builder, but configuration issues at the Kafka SinkConnectorConfig and SourceConnectorConfig interface layers make this difficult to implement, and make the code more complex.
3. Duplication of configuration property definitions and access code.

[mkeskells]

Is this proposal covering when we want to add configuration for an external (non Aiven) component, e.g. a custom record grouper like in #297?

[Claudenw]

Comment on the initial argument. We do not need to extend the Kafka SourceConnectorConfig, SinkConnectorConfig, or ConnectorConfig. We do need to extend the Kafka AbstractConfig class. But the design for components that are mixed in across hierarchies still applies.

[mkeskells]

You are swapping inheritance for composition, but requiring the caller to know of the composition, which isnt needed
Java doesnt allow scala like traits with implementation(shame), but does allow default methods, with removes the code duplication for all of the forwarder methods, and allows the caller to be unaware of the modelling of the diffferent facets of how you have compartmentalised the configuration. Effectively you encapsulate the composition

  // delegate to CommonsConfigFragment method, use to support existing code.
    /**
     * @deprecated use getCommonsConfig().getgetKafkaRetryBackoffMs()
     */
     @Deprecated
    public Long getKafkaRetryBackoffMs() {
        return commonConfig.getKafkaRetryBackoffMs();
    }

change public final class S3SinkConfig extends SinkCommonConfig { to public final class S3SinkConfig extends SinkCommonConfig implements S3AuthenticationAccess {
and have

public interface S3AuthenticationAccess {

     S3AuthenticationFragment getAuthFragment();
     
     default public Long getKafkaRetryBackoffMs() {
        return getAuthFragment().getKafkaRetryBackoffMs();
    }

Now you have the sources and sinks encapsulating all of the configuration access methods for all of the composed configuration fragments.

You could probably move the whole of the config to the interface, as the definion looks to be just static methods, that way the composition is just by assembly of the interfaces ( and the config init/validate etc which would have to be explicit as we dont have a real trait mixin :-) ), but you still have to do this in the existing proposal

I am not sure what the use case is for the fluent access to the common code fragment. IF thats needed then again it looks to require the caller to understand the composition, which I dont see a need for, and I think shoul dbe avoided. I dont know all of your code though ....