/**
* A condition on ConnectRecords.
* Implementations of this interface can be used for filtering records and conditionally applying Transformations.
* Implementations must be public and have a public constructor with no parameters.
*/
public Predicate<R extends ConnectRecord<R>> implements Configurable, Closeable {

    /** 
     * Configuration specification for this predicate. 
     */
    ConfigDef config();

    /**
     * Validate the predicate configuration values against configuration definitions.
     * @param predicateConfigs the provided configuration values
     * @return List of Config, each Config contains the updated configuration information given
     * the current configuration values.
     */
    default Config validate(Map<String, String> predicateConfigs) {
        ConfigDef configDef = config();
        if (null == configDef) {
            throw new ConnectException(
                String.format("%s.config() must return a ConfigDef that is not null.", this.getClass().getName())
            );
        }
        List<ConfigValue> configValues = configDef.validate(predicateConfigs);
        return new Config(configValues);
    }

    /**
     * Returns whether the given record satisfies this predicate.
     */
    boolean test(R record);

    @Override
    void close();
}

transforms: filter
transforms.filter.type: Filter
transforms.filter.condition?type: org.apache.kafka.connect.transforms.predicates.TopicNameMatch
transforms.filter.condition.?pattern: foo|bar
transforms.filter.?negate: false