diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnector.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnector.java index 548f52008..34e798918 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnector.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnector.java @@ -115,7 +115,8 @@ public List> taskConfigs(final int maxTasks) { try { Thread.sleep(THREAD_SLEEP_INTERVAL_MS); } catch (InterruptedException ex) { - // Action may be interrupted + Thread.currentThread().interrupt(); + break; } } diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfig.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfig.java index f645803f8..11d024606 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfig.java +++ b/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, diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfigVariables.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfigVariables.java index b69f679d5..65729ba7f 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkConnectorConfigVariables.java +++ b/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; } -} \ No newline at end of file +} diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkTask.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkTask.java index 1f345c896..f3ff627fe 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkTask.java +++ b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/ClickHouseSinkTask.java @@ -229,10 +229,14 @@ public void put(Collection 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); + } } } diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/config/DefaultColumnDataTypeMappingConfig.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/config/DefaultColumnDataTypeMappingConfig.java index bff0cec3c..2a4bd06a1 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/config/DefaultColumnDataTypeMappingConfig.java +++ b/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; diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/config/SchemaOverrideConfig.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/config/SchemaOverrideConfig.java index 815a3f327..78e84eada 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/config/SchemaOverrideConfig.java +++ b/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; /** diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/ColumnOverrides.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/ColumnOverrides.java index 5e9898aca..bac45bb7d 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/db/ColumnOverrides.java +++ b/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 columnOverridesMap = new HashMap<>(); + static Map 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 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; diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/validators/DatabaseOverrideValidator.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/validators/DatabaseOverrideValidator.java index e20490420..8f0b449c4 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/validators/DatabaseOverrideValidator.java +++ b/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: :,... Error: " + e.getMessage()); } } } diff --git a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/validators/TopicToTableValidator.java b/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/validators/TopicToTableValidator.java index c6ef239b8..aa1bdd9eb 100644 --- a/sink-connector/src/main/java/com/altinity/clickhouse/sink/connector/validators/TopicToTableValidator.java +++ b/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: :,:,..."); } } catch (Exception e) { - // Log the stack trace if an error occurs during validation - e.printStackTrace(); + throw new ConfigException(name, value, + "Format: :,:,... Error: " + e.getMessage()); } } diff --git a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/ColumnOverridesTest.java b/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/ColumnOverridesTest.java index a3dec89e7..547f8a31b 100644 --- a/sink-connector/src/test/java/com/altinity/clickhouse/sink/connector/db/ColumnOverridesTest.java +++ b/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"); } }