test: add test

Author: ekawinataa

dagger-common/build.gradle

@@ -139,11 +139,11 @@ publishing {
 protobuf {
     generatedFilesBaseDir = "$projectDir/src/generated"
     protoc {
-        artifact = "com.google.protobuf:protoc:3.1.0"
+        artifact = "com.google.protobuf:protoc:3.17.3"
     }
     plugins {
         grpc {
-            artifact = "io.grpc:protoc-gen-grpc-java:1.19.0"
+            artifact = "io.grpc:protoc-gen-grpc-java:1.60.1"
         }
     }
     generateProtoTasks {

dagger-core/src/main/java/com/gotocompany/dagger/core/sink/SinkOrchestrator.java

@@ -10,6 +10,7 @@
 import com.gotocompany.dagger.core.sink.influx.InfluxDBSink;
 import com.gotocompany.dagger.core.utils.KafkaConfigUtil;
 import com.gotocompany.dagger.core.utils.Constants;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.connector.sink.Sink;
 import org.apache.flink.api.java.utils.ParameterTool;
 import org.apache.flink.connector.base.DeliveryGuarantee;
@@ -24,6 +25,9 @@
 import com.gotocompany.dagger.core.sink.kafka.KafkaSerializerBuilder;
 import com.gotocompany.dagger.core.sink.log.LogSink;
 
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -117,6 +121,7 @@ protected Properties getProducerProperties(Configuration configuration) {
                 .map(ParameterTool::getProperties)
                 .orElseGet(Properties::new));
         kafkaProducerConfigs.putAll(dynamicProperties);
+        setSslPasswordsFromFile(configuration, kafkaProducerConfigs);
         return kafkaProducerConfigs;
     }
 
@@ -136,4 +141,23 @@ public Map<String, List<String>> getTelemetry() {
     private void addMetric(String key, String value) {
         metrics.computeIfAbsent(key, k -> new ArrayList<>()).add(value);
     }
+
+    private void setSslPasswordsFromFile(Configuration configuration, Properties kafkaProps) {
+        String sslTruststorePasswordFileLocation = configuration.getString(Constants.SINK_KAFKA_SSL_TRUSTSTORE_PASSWORD_LOCATION_KEY, StringUtils.EMPTY);
+        if (StringUtils.isNotEmpty(sslTruststorePasswordFileLocation)) {
+            kafkaProps.setProperty(Constants.KAFKA_PROPS_SSL_TRUSTSTORE_PASSWORD_KEY, parsePasswordFile(sslTruststorePasswordFileLocation));
+        }
+        String sslKeystorePasswordFileLocation = configuration.getString(Constants.SINK_KAFKA_SSL_KEYSTORE_PASSWORD_LOCATION_KEY, StringUtils.EMPTY);
+        if (StringUtils.isNotEmpty(sslKeystorePasswordFileLocation)) {
+            kafkaProps.setProperty(Constants.KAFKA_PROPS_SSL_TRUSTSTORE_PASSWORD_KEY, parsePasswordFile(sslKeystorePasswordFileLocation));
+        }
+    }
+
+    private String parsePasswordFile(String path) {
+        try {
+            return new String(Files.readAllBytes(Paths.get(path)));
+        } catch (IOException e) {
+            throw new IllegalArgumentException("Error reading password file: " + path, e);
+        }
+    }
 }

dagger-core/src/test/java/com/gotocompany/dagger/core/source/config/StreamConfigTest.java

@@ -267,6 +267,38 @@ public void shouldParseSslPasswordsConfig() throws IOException {
         assertEquals("truststore-password", streamProperties.getProperty("ssl.truststore.password"));
     }
 
+    @Test
+    public void shouldOverrideExplicitSslPasswordConfig() throws IOException {
+        File keystorePassword = writeDummyPasswordFile("ssl-keystore-password.txt", "keystore-password");
+        File truststorePassword = writeDummyPasswordFile("ssl-truststore-password.txt", "truststore-password");
+        when(configuration.getString(INPUT_STREAMS, ""))
+                .thenReturn(String.format("[{\"SOURCE_KAFKA_CONSUMER_CONFIG_SSL_KEY_PASSWORD\": \"password\", \"SOURCE_KAFKA_CONSUMER_CONFIG_SSL_KEYSTORE_PASSWORD_FILE_LOCATION\":  \"%s\", \"SOURCE_KAFKA_CONSUMER_CONFIG_SSL_TRUSTSTORE_PASSWORD_FILE_LOCATION\": \"%s\"}]", keystorePassword.getAbsolutePath(), truststorePassword.getAbsolutePath()));
+        StreamConfig[] streamConfigs = StreamConfig.parse(configuration);
+
+        Properties streamProperties = streamConfigs[0].getKafkaProps(configuration);
+
+        assertEquals("keystore-password", streamProperties.getProperty("ssl.keystore.password"));
+        assertEquals("truststore-password", streamProperties.getProperty("ssl.truststore.password"));
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void shouldThrowIllegalArgumentExceptionWhenKeystorePasswordFileNotExists() {
+        when(configuration.getString(INPUT_STREAMS, ""))
+                .thenReturn(String.format("[{\"SOURCE_KAFKA_CONSUMER_CONFIG_SSL_KEYSTORE_PASSWORD_FILE_LOCATION\":  \"%s\"}]", "non-exist.txt"));
+        StreamConfig[] streamConfigs = StreamConfig.parse(configuration);
+
+        streamConfigs[0].getKafkaProps(configuration);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void shouldThrowIllegalArgumentExceptionWhenTruststorePasswordFileNotExists() {
+        when(configuration.getString(INPUT_STREAMS, ""))
+                .thenReturn(String.format("[{\"SOURCE_KAFKA_CONSUMER_CONFIG_SSL_TRUSTSTORE_PASSWORD_FILE_LOCATION\":  \"%s\"}]", "non-exist.txt"));
+        StreamConfig[] streamConfigs = StreamConfig.parse(configuration);
+
+        streamConfigs[0].getKafkaProps(configuration);
+    }
+
     @Test(expected = IllegalArgumentException.class)
     public void shouldThrowIllegalArgumentExceptionIfAdditionalKafkaPropsNotMatchingPrefix() {
         String streamConfig = "[ { \"SOURCE_KAFKA_TOPIC_NAMES\": \"test-topic\", \"INPUT_SCHEMA_TABLE\": \"data_stream\", \"INPUT_SCHEMA_PROTO_CLASS\": \"com.tests.TestMessage\", \"INPUT_SCHEMA_EVENT_TIMESTAMP_FIELD_INDEX\": \"41\", \"SOURCE_KAFKA_CONSUMER_CONFIG_BOOTSTRAP_SERVERS\": \"localhost:9092\",\"SOURCE_KAFKA_CONSUMER_CONFIG_SECURITY_PROTOCOL\": \"SASL_PLAINTEXT\",\"SOURCE_KAFKA_CONSUMER_CONFIG_SASL_MECHANISM\":\"SCRAM-SHA-512\",\"SOURCE_KAFKA_CONSUMER_CONFIG_SASL_JAAS_CONFIG\":\"org.apache.kafka.common.security.scram.ScramLoginModule required username=\\\"username\\\" password=\\\"password\\\";\", \"SOURCE_KAFKA_CONSUMER_CONFIG_AUTO_COMMIT_ENABLE\": \"false\", \"SOURCE_KAFKA_CONSUMER_CONFIG_AUTO_OFFSET_RESET\": \"latest\", \"SOURCE_KAFKA_CONSUMER_CONFIG_GROUP_ID\": \"dummy-consumer-group\", \"SOURCE_KAFKA_NAME\": \"local-kafka-stream\", \"SOURCE_KAFKA_CONSUMER_ADDITIONAL_CONFIGURATIONS\": {\"SOURCE_KAFKA_CONSUMER_CONFIG_SSL_KEYSTORE_KEY\": \"ssl_keystore_key\", \"SOURCE_KAFKA_CONSUMER_CONFIG_SSL_KEYSTORE_LOCATION\": \"ssl_keystore_location\"} }, {\"INPUT_SCHEMA_TABLE\": \"data_stream_1\", \"SOURCE_KAFKA_TOPIC_NAMES\": \"test-topic\", \"INPUT_DATATYPE\": \"JSON\", \"INPUT_SCHEMA_JSON_SCHEMA\": \"{ \\\"$schema\\\": \\\"https://json-schema.org/draft/2020-12/schema\\\", \\\"$id\\\": \\\"https://example.com/product.schema.json\\\", \\\"title\\\": \\\"Product\\\", \\\"description\\\": \\\"A product from Acme's catalog\\\", \\\"type\\\": \\\"object\\\", \\\"properties\\\": { \\\"id\\\": { \\\"description\\\": \\\"The unique identifier for a product\\\", \\\"type\\\": \\\"string\\\" }, \\\"time\\\": { \\\"description\\\": \\\"event timestamp of the event\\\", \\\"type\\\": \\\"string\\\", \\\"format\\\" : \\\"date-time\\\" } }, \\\"required\\\": [ \\\"id\\\", \\\"time\\\" ] }\", \"INPUT_SCHEMA_EVENT_TIMESTAMP_FIELD_INDEX\": \"41\", \"SOURCE_KAFKA_CONSUMER_CONFIG_BOOTSTRAP_SERVERS\": \"localhost:9092\", \"SOURCE_KAFKA_CONSUMER_CONFIG_AUTO_COMMIT_ENABLE\": \"true\", \"SOURCE_KAFKA_CONSUMER_CONFIG_AUTO_OFFSET_RESET\": \"latest\", \"SOURCE_KAFKA_CONSUMER_CONFIG_GROUP_ID\": \"dummy-consumer-group\", \"SOURCE_KAFKA_NAME\": \"local-kafka-stream\", \"SOURCE_KAFKA_CONSUMER_ADDITIONAL_CONFIGURATIONS\": {\"SOURCE_KAFKA_CONSUMER_CONFIG_SSL_KEYSTORE_KEY\": \"ssl_keystore_key_2\", \"CONSUMER_CONFIG_SSL_KEYSTORE_LOCATION\": \"ssl_keystore_location_2\", \"SOURCE_KAFKA_CONSUMER_CONFIG_OFFSET_FLUSH_INTERVAL_MS\":\"1000\"} } ]";