better error messages

Author: jerrinot

connector/pom.xml

@@ -120,7 +120,7 @@
                             <goal>kafka-connect</goal>
                         </goals>
                         <configuration>
-                            <title>Kafka Connect QuestDB</title>
+                            <title>QuestDB Kafka Connector</title>
                             <documentationUrl>https://questdb.io/docs/third-party-tools/kafka/questdb-kafka/</documentationUrl>
                             <description>
                                 QuestDB Sink connector for Apache Kafka. QuestDB is a columnar time-series database with

connector/src/main/java/io/questdb/kafka/QuestDBSinkConnectorConfig.java

@@ -8,7 +8,6 @@
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.connect.errors.ConnectException;
 
-import java.time.format.DateTimeFormatter;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
@@ -106,7 +105,7 @@ public String getTable() {
         return getString(TABLE_CONFIG);
     }
 
-    public String getDefaultTimestampFormat() {
+    public String getTimestampFormat() {
         return getString(TIMESTAMP_FORMAT);
     }
 

connector/src/main/java/io/questdb/kafka/QuestDBSinkTask.java

@@ -61,7 +61,7 @@ public void start(Map<String, String> map) {
             for (String symbolColumn : timestampStringFields.split(",")) {
                 stringTimestampColumns.add(symbolColumn.trim());
             }
-            dataFormat = TimestampParserCompiler.compilePattern(config.getDefaultTimestampFormat());
+            dataFormat = TimestampParserCompiler.compilePattern(config.getTimestampFormat());
         } else {
             stringTimestampColumns = Collections.emptySet();
         }
@@ -314,7 +314,9 @@ private long parseToMicros(String timestamp) {
         try {
             return dataFormat.parse(timestamp, DateFormatUtils.enLocale);
         } catch (NumericException e) {
-            throw new ConnectException("Cannot parse timestamp: " + timestamp, e);
+            throw new ConnectException("Cannot parse timestamp: " + timestamp + " with the configured format '" + config.getTimestampFormat() +"' use '"
+                    + QuestDBSinkConnectorConfig.TIMESTAMP_FORMAT + "' to configure the right timestamp format. " +
+                    "See https://questdb.io/docs/reference/function/date-time/#date-and-timestamp-format for timestamp parser documentation. ", e);
         }
     }