Tighten constraints on TopicWiseConfigs

Currently if any config option starts with `topic.` prefix,
the connector will attempt to match it against regular expression
for TopicWiseConfigs. If it fails then whole connector fails to start.
This is a problem when launching on Confluent Cloud which adds
for example `topic.creation.default.max.message.bytes` config.

It mistakenly gets through the first check because it starts with `topic.`
and then trips up on regular expression. It seems there could be many more
configs like this that pertain to Kafka topic but are not this connectors
TopicWiseConfigs.

Adding suffix check should be sufficient and it should not break existing
configurations as long as those adhered to the existing documentation.

Author: Bouncheck

src/main/java/io/connect/scylladb/ScyllaDbSinkConnectorConfig.java

@@ -25,6 +25,8 @@
 import io.confluent.kafka.connect.utils.config.ValidEnum;
 import io.confluent.kafka.connect.utils.config.ValidPort;
 import io.connect.scylladb.topictotable.TopicConfigs;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Configuration class for {@link ScyllaDbSinkConnector}.
@@ -62,6 +64,10 @@ public class ScyllaDbSinkConnectorConfig extends AbstractConfig {
   private static final Pattern TOPIC_KS_TABLE_SETTING_PATTERN =
           Pattern.compile("topic\\.([a-zA-Z0-9._-]+)\\.([^.]+|\"[\"]+\")\\.([^.]+|\"[\"]+\")\\.(mapping|consistencyLevel|ttlSeconds|deletesEnabled)$");
 
+  private static final String[] TOPIC_WISE_CONFIGS_VALID_SUFFIXES = {".mapping",".consistencyLevel",".ttlSeconds",".deletesEnabled"};
+
+  private static final Logger log = LoggerFactory.getLogger(ScyllaDbSinkConnectorConfig.class);
+
   static final Set<String> CLIENT_COMPRESSION = ImmutableSet.of("none", "lz4", "snappy");
 
   static final Set<String> TABLE_COMPRESSORS = ImmutableSet.of("SnappyCompressor", "LZ4Compressor", "DeflateCompressor", "none");
@@ -111,8 +117,9 @@ public ScyllaDbSinkConnectorConfig(Map<?, ?> originals) {
     Map<String, Map<String, String>> topicWiseConfigsMap = new HashMap<>();
     for (final Map.Entry<String, String> entry : ((Map<String, String>) originals).entrySet()) {
       final String name2 = entry.getKey();
-      if (name2.startsWith("topic.")) {
+      if (name2.startsWith("topic.") && hasTopicWiseConfigSuffix(name2)) {
         final String topicName = this.tryMatchTopicName(name2);
+        log.debug("Interpreting " + name2 + " as custom TopicWiseConfig for topic " + topicName);
         final Map<String, String> topicMap = topicWiseConfigsMap.computeIfAbsent(topicName, t -> new HashMap());
         topicMap.put(name2.split("\\.")[name2.split("\\.").length - 1], entry.getValue());
       }
@@ -567,6 +574,13 @@ private String tryMatchTopicName(final String name) {
     throw new IllegalArgumentException("The setting: " + name + " does not match topic.keyspace.table nor topic.codec regular expression pattern");
   }
 
+  private boolean hasTopicWiseConfigSuffix(final String name) {
+    for (String suffix : TOPIC_WISE_CONFIGS_VALID_SUFFIXES) {
+      if (name.endsWith(suffix)) return true;
+    }
+    return false;
+  }
+
   private static String[] toStringArray(Object[] arr){
     return Arrays.stream(arr).map(Object::toString).toArray(String[]::new);
   }