Update common code (#559)

Update common code
Minimal changes to other code
Added configuration documentation.

Author: Claudenw

azure-sink-connector/src/main/java/io/aiven/kafka/connect/azure/sink/AzureBlobSinkConfig.java

@@ -17,30 +17,25 @@
 package io.aiven.kafka.connect.azure.sink;
 
 import java.time.Duration;
-import java.time.ZoneId;
-import java.time.ZoneOffset;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.regex.Pattern;
-import java.util.stream.Collectors;
 
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigException;
 
 import io.aiven.kafka.connect.common.config.AivenCommonConfig;
 import io.aiven.kafka.connect.common.config.CompressionType;
-import io.aiven.kafka.connect.common.config.FixedSetRecommender;
 import io.aiven.kafka.connect.common.config.OutputField;
 import io.aiven.kafka.connect.common.config.OutputFieldEncodingType;
 import io.aiven.kafka.connect.common.config.OutputFieldType;
-import io.aiven.kafka.connect.common.config.TimestampSource;
-import io.aiven.kafka.connect.common.config.validators.FilenameTemplateValidator;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+@SuppressWarnings("PMD.UnusedPrivateMethod")
 public final class AzureBlobSinkConfig extends AivenCommonConfig {
     private static final Logger LOG = LoggerFactory.getLogger(AzureBlobSinkConfig.class);
     private static final String USER_AGENT_HEADER_FORMAT = "Azure Blob Sink/%s (GPN: Aiven;)";
@@ -72,12 +67,10 @@ public final class AzureBlobSinkConfig extends AivenCommonConfig {
 
     public static final String NAME_CONFIG = "name";
 
-    public static ConfigDef configDef() {
-        final ConfigDef configDef = new ConfigDef();
+    public static SinkCommonConfigDef configDef() {
+        final SinkCommonConfigDef configDef = new SinkCommonConfigDef(OutputFieldType.VALUE, CompressionType.NONE);
         addAzureConfigGroup(configDef);
         addFileConfigGroup(configDef);
-        addOutputFieldsFormatConfigGroup(configDef, OutputFieldType.VALUE);
-        addKafkaBackoffPolicy(configDef);
         addAzureRetryPolicies(configDef);
         addUserAgentConfig(configDef);
         return configDef;
@@ -125,7 +118,6 @@ private static void addAzureRetryPolicies(final ConfigDef configDef) {
     }
 
     private static void addFileConfigGroup(final ConfigDef configDef) {
-        int fileGroupCounter = 0;
         configDef.define(FILE_NAME_PREFIX_CONFIG, ConfigDef.Type.STRING, "", new ConfigDef.Validator() {
             @Override
             public void ensureValid(final String name, final Object value) {
@@ -137,79 +129,7 @@ public void ensureValid(final String name, final Object value) {
                 }
             }
         }, ConfigDef.Importance.MEDIUM, "The prefix to be added to the name of each file put on Azure Blob.",
-                GROUP_FILE, fileGroupCounter++, ConfigDef.Width.NONE, FILE_NAME_PREFIX_CONFIG);
-
-        configDef.define(FILE_NAME_TEMPLATE_CONFIG, ConfigDef.Type.STRING, null,
-                new FilenameTemplateValidator(FILE_NAME_TEMPLATE_CONFIG), ConfigDef.Importance.MEDIUM,
-                "The template for file names on Azure Blob. "
-                        + "Supports `{{ variable }}` placeholders for substituting variables. "
-                        + "Currently supported variables are `topic`, `partition`, and `start_offset` "
-                        + "(the offset of the first record in the file). "
-                        + "Only some combinations of variables are valid, which currently are:\n"
-                        + "- `topic`, `partition`, `start_offset`.",
-                GROUP_FILE, fileGroupCounter++, ConfigDef.Width.LONG, FILE_NAME_TEMPLATE_CONFIG);
-
-        final String supportedCompressionTypes = CompressionType.names()
-                .stream()
-                .map(f -> "'" + f + "'")
-                .collect(Collectors.joining(", "));
-        configDef.define(FILE_COMPRESSION_TYPE_CONFIG, ConfigDef.Type.STRING, CompressionType.NONE.name,
-                new ConfigDef.Validator() {
-                    @Override
-                    public void ensureValid(final String name, final Object value) {
-                        assert value instanceof String;
-                        final String valueStr = (String) value;
-                        if (!CompressionType.names().contains(valueStr)) {
-                            throw new ConfigException(FILE_COMPRESSION_TYPE_CONFIG, valueStr,
-                                    "supported values are: " + supportedCompressionTypes);
-                        }
-                    }
-                }, ConfigDef.Importance.MEDIUM,
-                "The compression type used for files put on Azure Blob. " + "The supported values are: "
-                        + supportedCompressionTypes + ".",
-                GROUP_FILE, fileGroupCounter++, ConfigDef.Width.NONE, FILE_COMPRESSION_TYPE_CONFIG,
-                FixedSetRecommender.ofSupportedValues(CompressionType.names()));
-
-        configDef.define(FILE_MAX_RECORDS, ConfigDef.Type.INT, 0, new ConfigDef.Validator() {
-            @Override
-            public void ensureValid(final String name, final Object value) {
-                assert value instanceof Integer;
-                if ((Integer) value < 0) {
-                    throw new ConfigException(FILE_MAX_RECORDS, value, "must be a non-negative integer number");
-                }
-            }
-        }, ConfigDef.Importance.MEDIUM,
-                "The maximum number of records to put in a single file. " + "Must be a non-negative integer number. "
-                        + "0 is interpreted as \"unlimited\", which is the default.",
-                GROUP_FILE, fileGroupCounter++, ConfigDef.Width.SHORT, FILE_MAX_RECORDS);
-
-        configDef.define(FILE_NAME_TIMESTAMP_TIMEZONE, ConfigDef.Type.STRING, ZoneOffset.UTC.toString(),
-                new ConfigDef.Validator() {
-                    @Override
-                    public void ensureValid(final String name, final Object value) {
-                        try {
-                            ZoneId.of(value.toString());
-                        } catch (final Exception e) { // NOPMD broad exception catched
-                            throw new ConfigException(FILE_NAME_TIMESTAMP_TIMEZONE, value, e.getMessage());
-                        }
-                    }
-                }, ConfigDef.Importance.LOW,
-                "Specifies the timezone in which the dates and time for the timestamp variable will be treated. "
-                        + "Use standard shot and long names. Default is UTC",
-                GROUP_FILE, fileGroupCounter++, ConfigDef.Width.SHORT, FILE_NAME_TIMESTAMP_TIMEZONE);
-
-        configDef.define(FILE_NAME_TIMESTAMP_SOURCE, ConfigDef.Type.STRING, TimestampSource.Type.WALLCLOCK.name(),
-                new ConfigDef.Validator() {
-                    @Override
-                    public void ensureValid(final String name, final Object value) {
-                        try {
-                            TimestampSource.Type.of(value.toString());
-                        } catch (final Exception e) { // NOPMD broad exception catched
-                            throw new ConfigException(FILE_NAME_TIMESTAMP_SOURCE, value, e.getMessage());
-                        }
-                    }
-                }, ConfigDef.Importance.LOW, "Specifies the timestamp variable source. Default is wall-clock.",
-                GROUP_FILE, fileGroupCounter, ConfigDef.Width.SHORT, FILE_NAME_TIMESTAMP_SOURCE);
+                GROUP_FILE, 50, ConfigDef.Width.NONE, FILE_NAME_PREFIX_CONFIG);
     }
 
     public AzureBlobSinkConfig(final Map<String, String> properties) {