Status

Current state: "Under Discussion"

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA: here [Change the link from KAFKA-1 to your own ticket]

Motivation

Kafka's ConfigDef is used for specifying the set of expected configurations throughout the clients, Kafka Connect, and Kafka Streams. For each configuration, you can specify the name, the type, the default value, the documentation, the group information, the order in the group, the width of the configuration value, the name suitable for display in the UI, validation logic the user may provide to perform single configuration validation, recommenders to get valid values for a configuration given the current configuration values etc. Kafka's AbstractConfig class is a convenient base class for components to extend and use to parse, validate, and access configurations that adhere to a specified ConfigDef. The AbstractConfig class holds both the original configuration that was provided as well as the parsed configurations.

KIP-297 was recently introduce to provide a way to use variables within configuration files and a ConfigProvider extension to replace these variables with values found outside the configuration file. This mechanism allows Connect, for example, to store secrets outside of a connector configuration and to insert those externally stored secrets into the configuration just prior to usage. However, components that use the KIP-297 ConfigProviders must manage those providers and explicitly replace the variables.

This proposal intends to enhance the AbstractConfig base class to support replacing variables in configurations just prior to parsing and validation. This simple change will make it very easy for client applications, Kafka Connect, and Kafka Streams to use shared code to easily incorporate externalized secrets and other variable replacements within their configurations. This proposal does not add other ConfigProvider implementations or change the behavior of existing methods.

Proposed Changes

This proposal will add one new constructor and one static utility method to the AbstractConfig class:

/**
 * Construct a configuration with a ConfigDef, the configuration properties, optional {@link ConfigProviders}.
 * @param definition the definition of the configurations
 * @param originals the name-value pairs of the configuration
 * @param configProviders the optional named {@link ConfigProvider} instances to resolve any variables in {@code originals}; may be null or empty
 * @param doLog whether the configurations should be logged
 */
public AbstractConfig(ConfigDef definition, Map<?, ?> originals, Map<String, ConfigProvider> configProviders, boolean doLog) {
...
}

/**
 * Utility method to construct a map of named {@link ConfigProvider} implementations from the set of configuration properties for those ConfigProvider implementations.
 * @param properties the properties defining one or more ConfigProvider implementations
 * @return the map of named {@link ConfigProvider} instances
 */
public static Map<String, ConfigDef> loadConfigProviders(Map<?, ?> properties) {
...
}

The new constructor is similar to an existing constructor, with a new Map<String, ConfigProvider> configProviders parameter that provides a map of named ConfigProvider instances. The constructor will first find all variables in the values of the originals configurations, attempt to resolve the variables using the named ConfigProviders, and then do the normal parsing and validation of the configurations. If the map of ConfigProviders can be empty or null, the constructor will skip the variable substitution step and will simply validate and parse the supplied configuration. 

Variables are defined by KIP-297 and have the form "${providerName:[path:]key}", where "providerName" is the name of a ConfigProvider, "path" is an optional string, and "key" is a required string. Per KIP-297, this variable is resolved by passing the "key" and optional "path" to a ConfigProvider with the specified name, and the result from the ConfigProvider is then used in place of the variable. Variables that cannot be resolved by the AbstractConfig constructor will be left unchanged in the configuration.

The map of named ConfigProviders requires the caller construct and configure the ConfigProvider instances. Components and applications can instantiate these instances using any technique or mechanism, but many components may wish to load them using a configuration file that describes one or more ConfigProvider implementations. The new static loadConfigProviders(...) method provides a convenient mechanism for instantiating a set of named ConfigProvider instances using the configuration file format described in KIP-297. The following is an example of such a configuration file that defines a file-based ConfigProvider instance named "file" and another notional ConfigProvider implementation named "vault":

...
config.providers=file,vault
config.providers.file.class=org.apache.kafka.connect.configs.FileConfigProvider
config.providers.file.other.prop=another value passed to the class
config.providers.vault.class=com.acme.configs.CustomVaultConfigProvider
config.providers.vault.host=XYZ
config.providers.vault.password=ABCDEFG

The loadConfigProviders(...) method will support variables within the provided properties as long as the variable references a named ConfigProvider defined earlier in the config.providers property. For example, the following example would be allowed and the config.providers.vault.propB would be resolved correctly (assuming the "file" ConfigProvider can indeed resolve the variable):

...
config.providers=file,vault
config.providers.file.class=org.apache.kafka.connect.configs.FileConfigProvider
config.providers.file.other.prop=another value passed to the class
config.providers.vault.class=com.acme.configs.CustomVaultConfigProvider
config.providers.vault.host=XYZ
config.providers.vault.password=${file:/path/to/secrets.properties:vault.secret.password}

Example

Consider a configuration that includes the following two properties:

foo.baz=/usr/temp/
foo.bar=${file:/path/to/variables.properties:foo.bar}

The first foo.baz property is a typical name-value pair commonly used in all Kafka configuration files. The foo.bar property has a value that is a KIP-297 variable of the form "${providerName:[path:]key}", where "providerName" is the name of a ConfigProvider, "path" is an optional string, and "key" is a required string. Per KIP-297, this variable is resolved by passing the "foo.bar" key and "/path/to/variables.properties" path to a ConfigProvider with the name "file", and the string returned from the ConfigProvider is then used in place of the variable.

An application or component can use or create a custom subclass of the AbstractConfig, and use this class to parse and validate the configurations in a configuration file. If the application or component does not provide any ConfigProviders, then the AbstractConfig class will simply parse the configuration as before without resolving the variable. If, however, the component or application does pass a ConfigProvider named "file" to the AbstractConfig constructor, the constructor will attempt to use the "file" ConfigProvider instance to resolve and replace the foo.bar variable value prior to parsing and validating the configuration.  

// Define the configurations
ConfigDef configDef = ...

// Load the configuration file into a map
String configFileName = ...
Map<String, String> props = Utils.propsToStringMap(Utils.loadProps(workerPropsFile));

// Set up the config provider
ConfigProvider fileConfigProvider = new FileConfigProvider();
fileConfigProvider.configure(...)
Map<String, ConfigProvider> configProviders = Collections.singletonMap("file", fileConfigProvider);

// Parse and validate the configuration, using the ConfigProviders to resolve any variables
AbstractConfig config = new AbstractConfig(configDef, props, configProviders, true);

This new constructor will first look for any variables in the supplied props object, find the ${file:/path/to/variables.properties:foo.bar} variable and attempt to resolve it with the FileConfigProvider instance by making a call similar to fileConfigProvider.get("/path/to/variables.properties", Collections.singletonSet("foo.bar"), and using the result as the new value for the foo.bar property in the configuration. The constructor will then parse and validate the configuration using the supplied ConfigDef as normal.

A different application could choose to dynamically load the ConfigProvider instances from a single property file that contained both the normal configuration and the ConfigProvider configuration:

foo.baz=/usr/temp/
foo.bar=${file:/path/to/variables.properties:foo.bar}

config.providers=file,vault
config.providers.file.class=org.apache.kafka.connect.configs.FileConfigProvider
config.providers.file.other.prop=another value passed to the class

This second application might be as follows:

// Define the configurations
ConfigDef configDef = ...

// Load the configuration file into a map
String configFileName = ...
Map<String, String> props = Utils.propsToStringMap(Utils.loadProps(workerPropsFile));

// Set up the config providers
Map<String, ConfigProvider> configProviders = AbstractConfig.loadConfigProviders(props);

// Parse and validate the configuration, using the ConfigProviders to resolve any variables
AbstractConfig config = new AbstractConfig(configDef, props, configProviders, true);

This use of the AbstractConfig constructor does the same as the previous example.

Compatibility, Deprecation, and Migration Plan

• No additional changes required for existing components and applications that use AbstractConfig. In order to make use of this feature, components and applications will have to use the new constructor. 

Rejected Alternatives

Several other designs were considered but ultimately rejected.

1. Change AbstractConfig to automatically resolve variables of the form specified in KIP-297. This was rejected because it would change the behavior of existing code and might cause unexpected effects.
2. Assume the ConfigProvider instances are also configured in the same configuration provided to the AbstractConfig constructor. This was rejected because it did not provide enough flexibility. For example, in Kafka Connect, the worker configurations can define ConfigProviders, whereas the connector configurations do not. Assuming the ConfigProviders are defined in the connector configuration would change the behavior of Connect.
3. Place the new logic in a separate class or as static methods. This is a viable option, but the current design was thought to be a bit more usable.