[split 15/22] config: disable.drop.truncate + ddl.schema.change.* settings, validator NPE fixes

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnector.java

@@ -115,7 +115,8 @@ public List<Map<String, String>> taskConfigs(final int maxTasks) {
             try {
                 Thread.sleep(THREAD_SLEEP_INTERVAL_MS);
             } catch (InterruptedException ex) {
-                // Action may be interrupted
+                Thread.currentThread().interrupt();
+                break;
             }
         }
 

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfig.java

@@ -484,7 +484,7 @@ static ConfigDef newConfigDef() {
                         Type.LONG,
                         30,
                         Importance.LOW,
-                        "The time in seconds to flush cached data",
+                        "The time in milliseconds to flush cached data",
                         CONFIG_GROUP_CONNECTOR_CONFIG,
                         3,
                         ConfigDef.Width.NONE,
@@ -733,7 +733,7 @@ static ConfigDef newConfigDef() {
                 .define(
                         ERROR_TABLE_NAME.toString(),
                         Type.STRING,
-                        ERROR_TABLE_NAME.toString(),
+                        DEFAULT_ERROR_TABLE,
                         Importance.LOW,
                         "Default table name for storing error records",
                         CONFIG_GROUP_CONNECTOR_CONFIG,
@@ -808,6 +808,28 @@ static ConfigDef newConfigDef() {
                         ConfigDef.Width.NONE,
                         ClickHouseSinkConnectorConfigVariables.REPLICATION_HISTORY_REPLICATION_LOG_ONLY.toString()
                 )
+                .define(
+                        ClickHouseSinkConnectorConfigVariables.DDL_SCHEMA_CHANGE_TIMEOUT_MS.toString(),
+                        Type.LONG,
+                        30000L,
+                        Importance.LOW,
+                        "Maximum time in milliseconds to wait for ALTER TABLE schema changes to become visible in system.columns before proceeding. Prevents race conditions where the batch insert thread reads stale column metadata after DDL execution.",
+                        CONFIG_GROUP_CONNECTOR_CONFIG,
+                        ORDER_3,
+                        ConfigDef.Width.NONE,
+                        ClickHouseSinkConnectorConfigVariables.DDL_SCHEMA_CHANGE_TIMEOUT_MS.toString()
+                )
+                .define(
+                        ClickHouseSinkConnectorConfigVariables.DDL_SCHEMA_CHANGE_POLL_INTERVAL_MS.toString(),
+                        Type.LONG,
+                        100L,
+                        Importance.LOW,
+                        "Polling interval in milliseconds between checks of system.columns when waiting for ALTER TABLE schema changes to become visible.",
+                        CONFIG_GROUP_CONNECTOR_CONFIG,
+                        ORDER_3,
+                        ConfigDef.Width.NONE,
+                        ClickHouseSinkConnectorConfigVariables.DDL_SCHEMA_CHANGE_POLL_INTERVAL_MS.toString()
+                )
                 .define(
                         ClickHouseSinkConnectorConfigVariables.DATABASE_HOSTNAME.toString(),
                         Type.STRING,

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfigVariables.java

@@ -106,7 +106,16 @@ public enum ClickHouseSinkConnectorConfigVariables {
 
     REPLICATION_HISTORY_REPLICATION_LOG_ONLY("replication.history.replication_log_only"),
 
-    DATABASE_HOSTNAME("database.hostname");
+    // Union of both sides. This PR adds the two ddl.schema.change.* settings;
+    // develop added disable.drop.truncate. Keeping only one side would drop a
+    // config key whose own code reads it.
+    DDL_SCHEMA_CHANGE_TIMEOUT_MS("ddl.schema.change.timeout.ms"),
+
+    DDL_SCHEMA_CHANGE_POLL_INTERVAL_MS("ddl.schema.change.poll.interval.ms"),
+
+    DATABASE_HOSTNAME("database.hostname"),
+
+    DISABLE_DROP_TRUNCATE("disable.drop.truncate");
 
 
 
@@ -120,4 +129,4 @@ public enum ClickHouseSinkConnectorConfigVariables {
     public String toString() {
         return this.label;
     }
-}
+}

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkTask.java

@@ -229,10 +229,14 @@ public void put(Collection<SinkRecord> records) {
             }
         }
 
-        try {
-            this.records.put(batch);
-        } catch (InterruptedException e) {
-            throw new RetriableException(e);
+        // Only enqueue non-empty batches — empty batches cause
+        // IndexOutOfBoundsException in processBatch/addRecordsToHistoryTable
+        if (!batch.isEmpty()) {
+            try {
+                this.records.put(batch);
+            } catch (InterruptedException e) {
+                throw new RetriableException(e);
+            }
         }
     }
 

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/config/DefaultColumnDataTypeMappingConfig.java

@@ -1,7 +1,5 @@
 package com.altinity.clickhouse.sink.connector.config;
 
-import org.yaml.snakeyaml.Yaml;
-import java.io.InputStream;
 import java.util.HashMap;
 import java.util.Map;
 

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/config/SchemaOverrideConfig.java

@@ -1,11 +1,5 @@
 package com.altinity.clickhouse.sink.connector.config;
 
-import com.alibaba.fastjson.JSONObject;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
-
-import java.io.IOException;
-import java.io.InputStream;
 import java.util.Map;
 
 /**

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/ColumnOverrides.java

@@ -1,7 +1,9 @@
 package com.altinity.clickhouse.sink.connector.db;
 
-import java.util.HashMap;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.TreeMap;
 
 /**
  * Class that maps overrides of column data types. This is done specifically
@@ -12,11 +14,17 @@ public class ColumnOverrides {
     /**
      * Map of specific data type strings to their overridden forms.
      */
-    static Map<String, String> columnOverridesMap = new HashMap<>();
+    static Map<String, String> columnOverridesMap;
 
     static {
+        // Use TreeMap sorted by descending key length so that longer/more-specific
+        // keys like "Nullable(DateTime" are checked before shorter keys like "DateTime".
+        // This prevents DateTime64 columns from incorrectly matching the "DateTime" key.
+        columnOverridesMap = new TreeMap<>(Comparator.comparingInt(String::length).reversed()
+                .thenComparing(Comparator.naturalOrder()));
         columnOverridesMap.put("DateTime", "String");
         columnOverridesMap.put("Nullable(DateTime", "Nullable(String)");
+        columnOverridesMap.put("DateTime", "String");
     }
 
     /**
@@ -40,9 +48,21 @@ public ColumnOverrides() {
      *         is found, or {@code null} if no override applies.
      */
     public static String getColumnOverride(String dataType) {
-        for (String key : columnOverridesMap.keySet()) {
+        if (dataType == null) {
+            return null;
+        }
+        for (Map.Entry<String, String> entry : columnOverridesMap.entrySet()) {
+            String key = entry.getKey();
             if (dataType.contains(key)) {
-                return columnOverridesMap.get(key);
+                // Do not override DateTime64 — only plain DateTime needs the
+                // String workaround.  DateTime64 is handled correctly by JDBC.
+                if (key.equals("DateTime") && dataType.contains("DateTime64")) {
+                    continue;
+                }
+                if (key.equals("Nullable(DateTime") && dataType.contains("DateTime64")) {
+                    continue;
+                }
+                return entry.getValue();
             }
         }
         return null;

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/validators/DatabaseOverrideValidator.java

@@ -32,7 +32,8 @@ public void ensureValid(String name, Object value) {
                 );
             }
         } catch (Exception e) {
-            e.printStackTrace();
+            throw new ConfigException(name, value,
+                    "Format: <src_database-1>:<destination_database-1>,...  Error: " + e.getMessage());
         }
     }
 }

sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/validators/TopicToTableValidator.java

@@ -52,8 +52,8 @@ public void ensureValid(String name, Object value) {
                         "Format: <topic-1>:<table-1>,<topic-2>:<table-2>,...");
             }
         } catch (Exception e) {
-            // Log the stack trace if an error occurs during validation
-            e.printStackTrace();
+            throw new ConfigException(name, value,
+                    "Format: <topic-1>:<table-1>,<topic-2>:<table-2>,...  Error: " + e.getMessage());
         }
     }
 

sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/ColumnOverridesTest.java

@@ -1,26 +1,62 @@
 package com.altinity.clickhouse.sink.connector.db;
 
-import com.clickhouse.data.ClickHouseDataType;
-import org.junit.Assert;
+import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Tests for {@link ColumnOverrides} to verify that DateTime64 columns
+ * are NOT incorrectly overridden to String.
+ */
 public class ColumnOverridesTest {
 
     @Test
-    public void testMapping() {
-        String dateTime64Type = "DateTime64(3)";
-        String dataTime64OverrideType = ColumnOverrides.getColumnOverride(dateTime64Type);
-        Assert.assertTrue(dataTime64OverrideType.equalsIgnoreCase("String"));
+    @DisplayName("DateTime column should be overridden to String")
+    public void testDateTimeIsOverridden() {
+        String result = ColumnOverrides.getColumnOverride("DateTime");
+        assertEquals("String", result, "DateTime should be overridden to String");
+    }
 
-        String nullableDateTime64Type = "Nullable(DateTime64)";
-        String nullableDataTime64OverrideType = ColumnOverrides.getColumnOverride(nullableDateTime64Type);
-        Assert.assertTrue(nullableDataTime64OverrideType.equalsIgnoreCase("Nullable(String)"));
+    @Test
+    @DisplayName("Nullable(DateTime) column should be overridden to Nullable(String)")
+    public void testNullableDateTimeIsOverridden() {
+        String result = ColumnOverrides.getColumnOverride("Nullable(DateTime)");
+        assertEquals("Nullable(String)", result, "Nullable(DateTime) should be overridden to Nullable(String)");
+    }
 
-        Assert.assertTrue(ColumnOverrides.getColumnOverride(ClickHouseDataType.DateTime.name()).equalsIgnoreCase("String"));
-        Assert.assertNull(ColumnOverrides.getColumnOverride(ClickHouseDataType.Decimal.name()));
+    @Test
+    @DisplayName("DateTime64 column should NOT be overridden")
+    public void testDateTime64NotOverridden() {
+        String result = ColumnOverrides.getColumnOverride("DateTime64(3)");
+        assertNull(result, "DateTime64 should NOT be overridden — JDBC handles it correctly");
+    }
 
+    @Test
+    @DisplayName("Nullable(DateTime64) column should NOT be overridden")
+    public void testNullableDateTime64NotOverridden() {
+        String result = ColumnOverrides.getColumnOverride("Nullable(DateTime64(6))");
+        assertNull(result, "Nullable(DateTime64) should NOT be overridden");
+    }
 
-        Assert.assertNull(ColumnOverrides.getColumnOverride(ClickHouseDataType.Int16.name()));
-        Assert.assertTrue(ColumnOverrides.getColumnOverride(ClickHouseDataType.DateTime32.name()).equalsIgnoreCase(ClickHouseDataType.String.name()));
+    @Test
+    @DisplayName("null dataType should return null without NPE")
+    public void testNullDataTypeReturnsNull() {
+        String result = ColumnOverrides.getColumnOverride(null);
+        assertNull(result, "null input should return null");
+    }
+
+    @Test
+    @DisplayName("String dataType should not match any override")
+    public void testStringNotOverridden() {
+        String result = ColumnOverrides.getColumnOverride("String");
+        assertNull(result, "String should not be overridden");
+    }
+
+    @Test
+    @DisplayName("DateTime with timezone should be overridden")
+    public void testDateTimeWithTimezoneOverridden() {
+        String result = ColumnOverrides.getColumnOverride("DateTime('UTC')");
+        assertEquals("String", result, "DateTime('UTC') should be overridden to String");
     }
 }