add config for ILP auth

Author: jerrinot

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

@@ -32,6 +32,15 @@ public final class QuestDBSinkConnectorConfig extends AbstractConfig {
     public static final String SYMBOL_COLUMNS_CONFIG = "symbols";
     private static final String SYMBOL_COLUMNS_DOC = "Comma separated list of columns that should be symbol type";
 
+    public static final String USERNAME = "username";
+    private static final String USERNAME_DOC = "Username for QuestDB ILP authentication";
+
+    public static final String TOKEN = "token";
+    private static final String TOKEN_DOC = "Token for QuestDB ILP authentication";
+
+    public static final String TLS = "tls";
+    public static final String TLS_DOC = "Use TLS for connecting to QuestDB";
+
     public QuestDBSinkConnectorConfig(ConfigDef config, Map<String, String> parsedConfig) {
         super(config, parsedConfig);
     }
@@ -49,7 +58,10 @@ public static ConfigDef conf() {
                 .define(SKIP_UNSUPPORTED_TYPES_CONFIG, Type.BOOLEAN, false, Importance.MEDIUM, SKIP_UNSUPPORTED_TYPES_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)
-                .define(SYMBOL_COLUMNS_CONFIG, Type.STRING, null, Importance.MEDIUM, SYMBOL_COLUMNS_DOC);
+                .define(SYMBOL_COLUMNS_CONFIG, Type.STRING, null, Importance.MEDIUM, SYMBOL_COLUMNS_DOC)
+                .define(USERNAME, Type.STRING, null, Importance.MEDIUM, USERNAME_DOC)
+                .define(TOKEN, Type.STRING, null, Importance.MEDIUM, TOKEN_DOC)
+                .define(TLS, Type.BOOLEAN, false, Importance.MEDIUM, TLS_DOC);
     }
 
     public String getHost() {
@@ -83,4 +95,16 @@ public boolean isIncludeKey() {
     public String getSymbolColumns() {
         return getString(SYMBOL_COLUMNS_CONFIG);
     }
+
+    public String getUsername() {
+        return getString(USERNAME);
+    }
+
+    public String getToken() {
+        return getString(TOKEN);
+    }
+
+    public boolean isTls() {
+        return getBoolean(TLS);
+    }
 }

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

@@ -46,7 +46,17 @@ public void start(Map<String, String> map) {
     }
 
     private Sender createSender() {
-        Sender rawSender = Sender.builder().address(config.getHost()).build();
+        Sender.LineSenderBuilder builder = Sender.builder().address(config.getHost());
+        if (config.isTls()) {
+            builder.enableTls();
+        }
+        if (config.getUsername() != null) {
+            if (config.getToken() == null) {
+                throw new ConnectException("Token is required when username is provided");
+            }
+            builder.enableAuth(config.getUsername()).authToken(config.getToken());
+        }
+        Sender rawSender = builder.build();
         String symbolColumns = config.getSymbolColumns();
         if (symbolColumns == null) {
             return rawSender;

readme.md

@@ -49,6 +49,9 @@ The connector supports following Options:
 | timestamp.field.name   | STRING  | pickup_time                                                 | N/A                | Designated timestamp field name                            |
 | include.key            | BOOLEAN | false                                                       | true               | Include message key in target table                        |
 | symbols                | STRING  | instrument,stock                                            | N/A                | Comma separated list of columns that should be symbol type |
+| user                   | STRING  | user1                                                       | admin              | User name for QuestDB. Used only when token is non-empty   |
+| token                  | STRING  | <sub>QgHCOyq35D5HocCMrUGJinEsjEscJlCp7FZQETH21Bw</sub>      | N/A                | Token for QuestDB authentication                           |
+| tls                    | BOOLEAN | true                                                        | false              | Use TLS for QuestDB connection                             |
 
 ## Supported serialization formats
 The connector does not do data deserialization on its own. It relies on Kafka Connect converters to deserialize data. It's been tested predominantly with JSON, but it should work with any converter, including Avro. Converters can be configured using `key.converter` and `value.converter` options, see the table above.