From fc7b6853b93649ffbb6d2df38f0ac1348220a2ec Mon Sep 17 00:00:00 2001
From: "jaegwon.seo" <162448493+wornjs@users.noreply.github.com>
Date: Thu, 1 Aug 2024 05:05:55 +0900
Subject: [PATCH] change generate kafka connect properties from env (#10545)

Co-authored-by: david-leifker <114954101+david-leifker@users.noreply.github.com>
---
 docker/kafka-setup/env_to_properties.py | 24 +++++++++++++++
 docker/kafka-setup/kafka-setup.sh       | 40 +------------------------
 2 files changed, 25 insertions(+), 39 deletions(-)
 create mode 100644 docker/kafka-setup/env_to_properties.py

diff --git a/docker/kafka-setup/env_to_properties.py b/docker/kafka-setup/env_to_properties.py
new file mode 100644
index 00000000000000..8d8b8c3cc7b59f
--- /dev/null
+++ b/docker/kafka-setup/env_to_properties.py
@@ -0,0 +1,24 @@
+import os
+import re
+import sys
+
+
+def env_to_properties(env_prefix: str, properties_file: str):
+    pattern = re.compile('(?<=[^_])_(?=[^_])')
+    props = {}
+
+    for (env_name, val) in os.environ.items():
+        if env_name.startswith(env_prefix):
+            raw_name = env_name[len(env_prefix):].lower()
+            prop_dot = '.'.join(pattern.split(raw_name))
+            props[prop_dot] = val
+
+    with open(properties_file, 'a') as f:
+        for k, v in props.items():
+            f.writelines(f'{k}={v}\n')
+
+
+if __name__ == '__main__':
+    env_prefix = sys.argv[1]
+    properties_file = sys.argv[2]
+    env_to_properties(env_prefix, properties_file)
diff --git a/docker/kafka-setup/kafka-setup.sh b/docker/kafka-setup/kafka-setup.sh
index 439ffb4d4d8295..392cca94666419 100755
--- a/docker/kafka-setup/kafka-setup.sh
+++ b/docker/kafka-setup/kafka-setup.sh
@@ -10,46 +10,8 @@ fi
 . kafka-config.sh
 
 echo "bootstrap.servers=$KAFKA_BOOTSTRAP_SERVER" > $CONNECTION_PROPERTIES_PATH
-echo "security.protocol=$KAFKA_PROPERTIES_SECURITY_PROTOCOL" >> $CONNECTION_PROPERTIES_PATH
 
-## Add support for SASL_PLAINTEXT
-if [[ $KAFKA_PROPERTIES_SECURITY_PROTOCOL == "SASL_PLAINTEXT" ]]; then
-    echo "sasl.mechanism=$KAFKA_PROPERTIES_SASL_MECHANISM" >> $CONNECTION_PROPERTIES_PATH
-    echo "sasl.jaas.config=$KAFKA_PROPERTIES_SASL_JAAS_CONFIG" >> $CONNECTION_PROPERTIES_PATH
-    echo "sasl.kerberos.service.name=$KAFKA_PROPERTIES_SASL_KERBEROS_SERVICE_NAME" >> $CONNECTION_PROPERTIES_PATH
-fi
-
-## Add support for SASL_SSL
-if [[ $KAFKA_PROPERTIES_SECURITY_PROTOCOL == "SASL_SSL" ]]; then
-    echo "sasl.jaas.config=$KAFKA_PROPERTIES_SASL_JAAS_CONFIG" >> $CONNECTION_PROPERTIES_PATH
-    echo "sasl.mechanism=$KAFKA_PROPERTIES_SASL_MECHANISM" >> $CONNECTION_PROPERTIES_PATH
-fi
-
-if [[ $KAFKA_PROPERTIES_SECURITY_PROTOCOL == "SSL" ]]; then
-    if [[ -n $KAFKA_PROPERTIES_SSL_KEYSTORE_LOCATION ]]; then
-        echo "ssl.keystore.location=$KAFKA_PROPERTIES_SSL_KEYSTORE_LOCATION" >> $CONNECTION_PROPERTIES_PATH
-        echo "ssl.keystore.password=$KAFKA_PROPERTIES_SSL_KEYSTORE_PASSWORD" >> $CONNECTION_PROPERTIES_PATH
-        echo "ssl.key.password=$KAFKA_PROPERTIES_SSL_KEY_PASSWORD" >> $CONNECTION_PROPERTIES_PATH
-        if [[ -n $KAFKA_PROPERTIES_SSL_KEYSTORE_TYPE ]]; then
-            echo "ssl.keystore.type=$KAFKA_PROPERTIES_SSL_KEYSTORE_TYPE" >> $CONNECTION_PROPERTIES_PATH
-        fi
-    fi
-    if [[ -n $KAFKA_PROPERTIES_SSL_TRUSTSTORE_LOCATION ]]; then
-        echo "ssl.truststore.location=$KAFKA_PROPERTIES_SSL_TRUSTSTORE_LOCATION" >> $CONNECTION_PROPERTIES_PATH
-        if [[ $KAFKA_PROPERTIES_SSL_TRUSTSTORE_TYPE != "PEM" ]]; then
-          echo "ssl.truststore.password=$KAFKA_PROPERTIES_SSL_TRUSTSTORE_PASSWORD" >> $CONNECTION_PROPERTIES_PATH
-        fi
-        if [[ -n $KAFKA_PROPERTIES_SSL_TRUSTSTORE_TYPE ]]; then
-            echo "ssl.truststore.type=$KAFKA_PROPERTIES_SSL_TRUSTSTORE_TYPE" >> $CONNECTION_PROPERTIES_PATH
-        fi
-    fi
-    echo "ssl.endpoint.identification.algorithm=$KAFKA_PROPERTIES_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM" >> $CONNECTION_PROPERTIES_PATH
-fi
-
-# Add support for SASL_CLIENT_CALLBACK_HANDLER_CLASS
-if [[ -n "$KAFKA_PROPERTIES_SASL_CLIENT_CALLBACK_HANDLER_CLASS" ]]; then
-    echo "sasl.client.callback.handler.class=$KAFKA_PROPERTIES_SASL_CLIENT_CALLBACK_HANDLER_CLASS" >> $CONNECTION_PROPERTIES_PATH
-fi
+python env_to_properties.py KAFKA_PROPERTIES_ $CONNECTION_PROPERTIES_PATH
 
 # cub kafka-ready -c $CONNECTION_PROPERTIES_PATH -b $KAFKA_BOOTSTRAP_SERVER 1 180
 . kafka-ready.sh