refactor(filters): cleanup classes

Author: fhussonnois

connect-file-pulse-filters/src/main/java/io/streamthoughts/kafka/connect/filepulse/config/CommonFilterConfig.java

@@ -111,7 +111,7 @@ public String source() {
         return getString(CommonFilterConfig.FILTER_SOURCE_FIELD_CONFIG);
     }
 
-    public static ConfigDef.ConfigKey withOverwrite(final String group, final int groupCounter) {
+    public static ConfigDef.ConfigKey getOverwriteConfigKey(final String group, final int groupCounter) {
         return new ConfigDef.ConfigKey(
                 CommonFilterConfig.FILTER_OVERWRITE_CONFIG,
                 ConfigDef.Type.LIST,

connect-file-pulse-filters/src/main/java/io/streamthoughts/kafka/connect/filepulse/config/ExplodeFilterConfig.java

@@ -39,7 +39,7 @@ public ExplodeFilterConfig(final Map<?, ?> originals) {
     public static ConfigDef configDef() {
         int filterGroupCounter = 0;
         return new ConfigDef(CommonFilterConfig.configDef())
-                .define(withOverwrite(FILTER_EXPLODE, filterGroupCounter++))
+                .define(getOverwriteConfigKey(FILTER_EXPLODE, filterGroupCounter++))
                 .define(getSourceConfigKey(FILTER_EXPLODE, filterGroupCounter++));
     }
 }

connect-file-pulse-filters/src/main/java/io/streamthoughts/kafka/connect/filepulse/config/GrokFilterConfig.java

@@ -46,7 +46,7 @@ public static ConfigDef configDef() {
         int filterGroupCounter = 0;
         final ConfigDef def = new ConfigDef(CommonFilterConfig.configDef())
                 .define(getSourceConfigKey(GROK_FILTER, filterGroupCounter++))
-                .define(withOverwrite(GROK_FILTER, filterGroupCounter++));
+                .define(getOverwriteConfigKey(GROK_FILTER, filterGroupCounter++));
         for (ConfigDef.ConfigKey configKey : GrokConfig.configDef().configKeys().values()) {
             def.define(new ConfigDef.ConfigKey(
                     configKey.name,

connect-file-pulse-filters/src/main/java/io/streamthoughts/kafka/connect/filepulse/config/JSONFilterConfig.java

@@ -76,7 +76,7 @@ public Charset charset() {
     public static ConfigDef configDef() {
         int filterGroupCounter = 0;
         return new ConfigDef(CommonFilterConfig.configDef())
-                .define(withOverwrite(JSON_FILTER, filterGroupCounter++))
+                .define(getOverwriteConfigKey(JSON_FILTER, filterGroupCounter++))
                 .define(getSourceConfigKey(JSON_FILTER, filterGroupCounter++))
                 .define(
                         JSON_TARGET_CONFIG,

connect-file-pulse-filters/src/main/java/io/streamthoughts/kafka/connect/filepulse/config/MoveFilterConfig.java

@@ -58,7 +58,7 @@ public String source() {
     public static ConfigDef configDef() {
         int filterGroupCounter = 0;
         return new ConfigDef(CommonFilterConfig.configDef())
-                .define(withOverwrite(GROUP_MOVE, filterGroupCounter++))
+                .define(getOverwriteConfigKey(GROUP_MOVE, filterGroupCounter++))
                 .define(getSourceConfigKey(GROUP_MOVE, filterGroupCounter++))
                 .define(
                         MOVE_TARGET_CONFIG,

connect-file-pulse-filters/src/main/java/io/streamthoughts/kafka/connect/filepulse/config/XmlToJsonFilterConfig.java

@@ -55,7 +55,7 @@ public XmlToJsonFilterConfig(final Map<?, ?> originals) {
     public static ConfigDef configDef() {
         int filterGroupCounter = 0;
         return new ConfigDef(CommonFilterConfig.configDef())
-                .define(withOverwrite(FILTER, filterGroupCounter++))
+                .define(getOverwriteConfigKey(FILTER, filterGroupCounter++))
                 .define(getSourceConfigKey(FILTER, filterGroupCounter++))
                 .define(
                         XML_PARSER_KEEP_STRINGS_CONFIG,

connect-file-pulse-filters/src/main/java/io/streamthoughts/kafka/connect/filepulse/filter/AbstractRecordFilter.java

@@ -36,7 +36,7 @@ public abstract class AbstractRecordFilter<T extends AbstractRecordFilter<T>> im
 
     private boolean ignoreFailure;
 
-    private CommonFilterConfig config;
+    protected CommonFilterConfig config;
 
     /**
      * {@inheritDoc}
@@ -58,12 +58,15 @@ public void configure(final Map<String, ?> configs) {
     @Override
     public void configure(final Map<String, ?> props, final RecordFilterProvider provider) {
         config = new CommonFilterConfig(props);
-        condition = config.condition();
-        ignoreFailure = config.ignoreFailure();
+        withOnCondition(config.condition());
+        withIgnoreFailure(config.ignoreFailure());
         if (!config.onFailure().isEmpty()) {
-            failurePipeline = new DefaultRecordFilterPipeline(config.onFailure().stream()
+            withOnFailure(
+                new DefaultRecordFilterPipeline(config.onFailure().stream()
                     .map(provider::getRecordForAlias)
-                    .collect(Collectors.toList()));
+                    .collect(Collectors.toList())
+                )
+            );
         }
         configure(props);
     }