option to exclude key from target table table

Author: jerrinot

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

@@ -23,9 +23,12 @@ public final class QuestDBSinkConnectorConfig extends AbstractConfig {
     public static final String SKIP_UNSUPPORTED_TYPES_CONFIG = "skip.unsupported.types";
     private static final String SKIP_UNSUPPORTED_TYPES_DOC = "Skip unsupported types";
 
-    public static final String DESIGNATED_TIMESTAMP_COLUMN_NAME = "timestamp.field.name";
+    public static final String DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG = "timestamp.field.name";
     private static final String DESIGNATED_TIMESTAMP_COLUMN_NAME_DOC = "Designated timestamp field name";
 
+    public static final String INCLUDE_KEY_CONFIG = "include.key";
+    private static final String INCLUDE_KEY_DOC = "Include key in the table";
+
     public QuestDBSinkConnectorConfig(ConfigDef config, Map<String, String> parsedConfig) {
         super(config, parsedConfig);
     }
@@ -41,7 +44,8 @@ public static ConfigDef conf() {
                 .define(KEY_PREFIX_CONFIG, Type.STRING, "key", Importance.MEDIUM, KEY_PREFIX_DOC)
                 .define(VALUE_PREFIX_CONFIG, Type.STRING, "", Importance.MEDIUM, VALUE_PREFIX_DOC)
                 .define(SKIP_UNSUPPORTED_TYPES_CONFIG, Type.BOOLEAN, false, Importance.MEDIUM, SKIP_UNSUPPORTED_TYPES_DOC)
-                .define(DESIGNATED_TIMESTAMP_COLUMN_NAME, Type.STRING, null, Importance.MEDIUM, DESIGNATED_TIMESTAMP_COLUMN_NAME_DOC);
+                .define(DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG, Type.STRING, null, Importance.MEDIUM, DESIGNATED_TIMESTAMP_COLUMN_NAME_DOC)
+                .define(INCLUDE_KEY_CONFIG, Type.BOOLEAN, true, Importance.MEDIUM, INCLUDE_KEY_DOC);
     }
 
     public String getHost() {
@@ -65,6 +69,10 @@ public boolean isSkipUnsupportedTypes() {
     }
 
     public String getDesignatedTimestampColumnName() {
-        return getString(DESIGNATED_TIMESTAMP_COLUMN_NAME);
+        return getString(DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG);
+    }
+
+    public boolean isIncludeKey() {
+        return getBoolean(INCLUDE_KEY_CONFIG);
     }
 }

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

@@ -59,7 +59,9 @@ private void handleSingleRecord(SinkRecord record) {
         sender.table(tableName);
 
         //todo: detect duplicated columns
-        handleObject(config.getKeyPrefix(), record.keySchema(), record.key(), PRIMITIVE_KEY_FALLBACK_NAME);
+        if (config.isIncludeKey()) {
+            handleObject(config.getKeyPrefix(), record.keySchema(), record.key(), PRIMITIVE_KEY_FALLBACK_NAME);
+        }
         handleObject(config.getValuePrefix(), record.valueSchema(), record.value(), PRIMITIVE_VALUE_FALLBACK_NAME);
 
         if (timestampColumnValue == null) {

connector/src/test/java/io/questdb/kafka/QuestDBSinkConnectorEmbeddedTest.java

@@ -40,7 +40,6 @@
 import static java.util.Collections.singletonMap;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.SECONDS;
-import static org.apache.kafka.common.config.AbstractConfig.CONFIG_PROVIDERS_CONFIG;
 import static org.apache.kafka.connect.runtime.ConnectorConfig.KEY_CONVERTER_CLASS_CONFIG;
 import static org.apache.kafka.connect.runtime.ConnectorConfig.VALUE_CONVERTER_CLASS_CONFIG;
 
@@ -63,7 +62,7 @@ public final class QuestDBSinkConnectorEmbeddedTest {
             .withEnv("JAVA_OPTS", "-Djava.locale.providers=JRE,SPI");
 
     @BeforeEach
-    public void setUp() throws IOException {
+    public void setUp() {
         topicName = newTopicName();
         JsonConverter jsonConverter = new JsonConverter();
         jsonConverter.configure(singletonMap(ConverterConfig.TYPE_CONFIG, ConverterType.VALUE.getName()));
@@ -120,7 +119,7 @@ public void testDesignatedTimestamp_noSchema_unixEpochMillis() {
         connect.kafka().createTopic(topicName, 1);
         Map<String, String> props = baseConnectorProps(topicName);
         props.put("value.converter.schemas.enable", "false");
-        props.put(QuestDBSinkConnectorConfig.DESIGNATED_TIMESTAMP_COLUMN_NAME, "birth");
+        props.put(QuestDBSinkConnectorConfig.DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG, "birth");
         connect.configureConnector(CONNECTOR_NAME, props);
         assertConnectorTaskRunningEventually();
 
@@ -141,7 +140,7 @@ public void testDesignatedTimestamp_noSchema_dateTransform_fromStringToTimestamp
         props.put("transforms.convert_birth.target.type", "Timestamp");
         props.put("transforms.convert_birth.field", "birth");
         props.put("transforms.convert_birth.format", "yyyy-MM-dd'T'HH:mm:ss.SSSX");
-        props.put(QuestDBSinkConnectorConfig.DESIGNATED_TIMESTAMP_COLUMN_NAME, "birth");
+        props.put(QuestDBSinkConnectorConfig.DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG, "birth");
         connect.configureConnector(CONNECTOR_NAME, props);
         assertConnectorTaskRunningEventually();
 
@@ -156,7 +155,7 @@ public void testDesignatedTimestamp_noSchema_dateTransform_fromStringToTimestamp
     public void testDesignatedTimestamp_withSchema() {
         connect.kafka().createTopic(topicName, 1);
         Map<String, String> props = baseConnectorProps(topicName);
-        props.put(QuestDBSinkConnectorConfig.DESIGNATED_TIMESTAMP_COLUMN_NAME, "birth");
+        props.put(QuestDBSinkConnectorConfig.DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG, "birth");
         connect.configureConnector(CONNECTOR_NAME, props);
         assertConnectorTaskRunningEventually();
         Schema schema = SchemaBuilder.struct().name("com.example.Person")
@@ -184,6 +183,39 @@ public void testDesignatedTimestamp_withSchema() {
                 "select * from " + topicName);
     }
 
+    @Test
+    public void testDoNotIncludeKey() {
+        connect.kafka().createTopic(topicName, 1);
+        Map<String, String> props = baseConnectorProps(topicName);
+        props.put(QuestDBSinkConnectorConfig.DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG, "birth");
+        props.put(QuestDBSinkConnectorConfig.INCLUDE_KEY_CONFIG, "false");
+        connect.configureConnector(CONNECTOR_NAME, props);
+        assertConnectorTaskRunningEventually();
+        Schema schema = SchemaBuilder.struct().name("com.example.Person")
+                .field("firstname", Schema.STRING_SCHEMA)
+                .field("lastname", Schema.STRING_SCHEMA)
+                .field("birth", Timestamp.SCHEMA)
+                .build();
+
+        java.util.Date birth = new Calendar.Builder()
+                .setTimeZone(TimeZone.getTimeZone("UTC"))
+                .setDate(2022, 9, 23) // note: month is 0-based
+                .setTimeOfDay(13, 53, 59, 123)
+                .build().getTime();
+
+        Struct struct = new Struct(schema)
+                .put("firstname", "John")
+                .put("lastname", "Doe")
+                .put("birth", birth);
+
+
+        connect.kafka().produce(topicName, "key", new String(converter.fromConnectData(topicName, schema, struct)));
+
+        QuestDBUtils.assertSqlEventually(questDBContainer, "\"firstname\",\"lastname\",\"timestamp\"\r\n"
+                        + "\"John\",\"Doe\",\"2022-10-23T13:53:59.123000Z\"\r\n",
+                "select * from " + topicName);
+    }
+
     @Test
     public void testJsonNoSchema() {
         connect.kafka().createTopic(topicName, 1);