Added unit test case and logic to limit the Date values to clickhouse supported date ranges.

doc/Monitoring.md

@@ -8,3 +8,5 @@ JMX metrics of sink connector are exposed through the port
 The JMX_exporter docker image scrapes the JMX metrics from the sink connector
 The metrics can be read through the following URL
 http://localhost:9072/metrics
+
+A Grafana dashboard is included to view JMX metrics.

src/main/java/com/altinity/clickhouse/sink/connector/converters/DebeziumConverter.java

@@ -1,5 +1,7 @@
 package com.altinity.clickhouse.sink.connector.converters;
 
+import com.altinity.clickhouse.sink.connector.db.Constants;
+
 import java.sql.Date;
 import java.text.SimpleDateFormat;
 import java.time.Instant;
@@ -45,7 +47,11 @@ public class DebeziumConverter {
 
     public static class DateConverter {
 
+
         /**
+         * MySQL: The DATE type is used for values with a date part but no time part.
+         * MySQL retrieves and displays DATE values in 'YYYY-MM-DD' format. The supported range is '1000-01-01' to '9999-12-31'.
+         *
          * Function to convert Debezium Date fields
          * to java.sql.Date
          * @param value
@@ -54,7 +60,24 @@ public class DebeziumConverter {
         public static Date convert(Object value) {
             long msSinceEpoch = TimeUnit.DAYS.toMillis((Integer) value);
             java.util.Date date = new java.util.Date(msSinceEpoch);
-            return new java.sql.Date(date.getTime());
+
+            java.util.Date modifiedDate = checkIfDateExceedsSupportedRange(date);
+
+            return new java.sql.Date(modifiedDate.getTime());
+        }
+
+        public static java.util.Date checkIfDateExceedsSupportedRange(java.util.Date providedDate) {
+            java.util.Date minSupportedDate = Date.valueOf(Constants.CLICKHOUSE_MIN_SUPPORTED_DATE);
+            java.util.Date maxSupportedDate = Date.valueOf(Constants.CLICKHOUSE_MAX_SUPPORTED_DATE);
+
+            if(providedDate.before(minSupportedDate)) {
+                return minSupportedDate;
+            } else if (providedDate.after(maxSupportedDate)){
+                return maxSupportedDate;
+            }
+
+            return providedDate;
+
         }
     }
 

src/main/java/com/altinity/clickhouse/sink/connector/db/Constants.java

@@ -0,0 +1,7 @@
+package com.altinity.clickhouse.sink.connector.db;
+
+public class Constants
+{
+    public static final String CLICKHOUSE_MIN_SUPPORTED_DATE = "1970-01-01";
+    public static final String CLICKHOUSE_MAX_SUPPORTED_DATE = "2149-06-06";
+}

src/test/com/altinity/clickhouse/sink/connector/converters/DebeziumConverterTest.java

@@ -1,5 +1,6 @@
 package com.altinity.clickhouse.sink.connector.converters;
 
+import com.altinity.clickhouse.sink.connector.db.Constants;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -32,6 +33,23 @@ public class DebeziumConverterTest {
         Assert.assertTrue(formattedDate.toString().equalsIgnoreCase("1979-12-31"));
     }
 
+    @Test
+    public void testDateConverterMinRange() {
+
+        Integer date = -144450000;
+        java.sql.Date formattedDate = DebeziumConverter.DateConverter.convert(date);
+
+        Assert.assertTrue(formattedDate.toString().equalsIgnoreCase(Constants.CLICKHOUSE_MIN_SUPPORTED_DATE));
+    }
+    @Test
+    public void testDateConverterMaxRange() {
+
+        Integer date = 450000;
+        java.sql.Date formattedDate = DebeziumConverter.DateConverter.convert(date);
+
+        Assert.assertTrue(formattedDate.toString().equalsIgnoreCase(Constants.CLICKHOUSE_MAX_SUPPORTED_DATE));
+    }
+
     @Test
     public void testZonedTimestampConverter() {