feat: kafka enhance support kerberos

Author: jarad0628

connectors/kafka-enhanced-connector/src/main/java/io/tapdata/kafka/KafkaConfig.java

@@ -10,6 +10,7 @@
 import io.tapdata.kafka.constants.KafkaConcurrentReadMode;
 import io.tapdata.kafka.constants.KafkaSchemaMode;
 import io.tapdata.kafka.constants.KafkaSerialization;
+import io.tapdata.kafka.utils.Krb5Util;
 import io.tapdata.pdk.apis.context.TapConnectionContext;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.clients.CommonClientConfigs;
@@ -35,8 +36,11 @@ public class KafkaConfig extends BasicConfig implements
         IConnectionACL,
         ConnectionExtParams {
 
-    public KafkaConfig(TapConnectionContext context) {
+    private final String connectorId;
+
+    public KafkaConfig(TapConnectionContext context, String connectorId) {
         super(context);
+        this.connectorId = connectorId;
     }
 
     // ---------- 连接配置 ----------
@@ -207,6 +211,10 @@ private Properties buildProperties(String type) {
                     " username='" + getSaslUsername() +
                     "' password='" + getSaslPassword() + "';");
         }
+        if (useKerberos()) {
+            String krb5Path = Krb5Util.saveByCatalog("connections-" + connectorId, getKrb5Keytab(), getKrb5Conf(), true);
+            Krb5Util.updateKafkaConf(getKrb5ServiceName(), getKrb5Principal(), krb5Path, getKrb5Conf(), props);
+        }
         if (useSsl()) {
 //            ssl.truststore.location=/path/to/kafka.client.truststore.jks
 //            ssl.truststore.password=truststore_password
@@ -320,7 +328,7 @@ public Properties buildProducerConfig() {
 
     // ---------- 静态方法 ----------
 
-    public static KafkaConfig valueOf(TapConnectionContext context) {
-        return new KafkaConfig(context);
+    public static KafkaConfig valueOf(TapConnectionContext context, String connectorId) {
+        return new KafkaConfig(context, connectorId);
     }
 }

connectors/kafka-enhanced-connector/src/main/java/io/tapdata/kafka/KafkaEnhancedConnector.java

@@ -9,6 +9,7 @@
 import io.tapdata.kit.EmptyKit;
 import io.tapdata.pdk.apis.annotations.TapConnectorClass;
 import io.tapdata.pdk.apis.context.TapConnectionContext;
+import io.tapdata.pdk.apis.context.TapConnectorContext;
 import io.tapdata.pdk.apis.entity.ConnectionOptions;
 import io.tapdata.pdk.apis.entity.TestItem;
 import io.tapdata.pdk.apis.functions.ConnectorFunctions;
@@ -37,15 +38,18 @@ public class KafkaEnhancedConnector extends ConnectorBase {
     @Override
     public void onStart(TapConnectionContext connectionContext) throws Throwable {
         connectionContext.getLog().info("Starting {}", PDK_ID);
-        kafkaConfig = KafkaConfig.valueOf(connectionContext);
-        kafkaService = new KafkaService(kafkaConfig, stopping);
         isConnectorStarted(connectionContext, connectorContext -> {
             String firstConnectorId = (String) connectorContext.getStateMap().get("firstConnectorId");
             if (EmptyKit.isNull(firstConnectorId)) {
                 firstConnectorId = UUID.randomUUID().toString().replace("-", "");
                 connectorContext.getStateMap().put("firstConnectorId", firstConnectorId);
             }
+            kafkaConfig = KafkaConfig.valueOf(connectionContext, firstConnectorId);
         });
+        if (!(connectionContext instanceof TapConnectorContext)) {
+            kafkaConfig = KafkaConfig.valueOf(connectionContext, "");
+        }
+        kafkaService = new KafkaService(kafkaConfig, stopping);
     }
 
     @Override

connectors/kafka-enhanced-connector/src/main/java/io/tapdata/kafka/KafkaTester.java

@@ -17,7 +17,7 @@
  */
 public class KafkaTester extends Tester<KafkaConfig> {
     public KafkaTester(TapConnectionContext connectionContext, Consumer<TestItem> consumer) {
-        super(KafkaConfig.valueOf(connectionContext), consumer);
+        super(KafkaConfig.valueOf(connectionContext, ""), consumer);
     }
 
     @Override

connectors/kafka-enhanced-connector/src/main/java/io/tapdata/kafka/config/IConnectionSecurity.java

@@ -14,6 +14,7 @@ public interface IConnectionSecurity extends IConfigWithContext {
     String KEY_SASL_USERNAME = "saslUsername";
     String KEY_SASL_PASSWORD = "saslPassword";
     String KEY_USE_SSL = "useSsl";
+    String KEY_USE_KERBEROS = "krb5";
 
     enum Protocol {
         PLAINTEXT, // 没有任何 authentication
@@ -70,6 +71,26 @@ default boolean useSasl() {
         return connectionConfigGet(KEY_USE_SASL, false);
     }
 
+    default boolean useKerberos() {
+        return connectionConfigGet(KEY_USE_KERBEROS, false);
+    }
+
+    default String getKrb5Keytab() {
+        return connectionConfigGet("krb5Keytab", "");
+    }
+
+    default String getKrb5Conf() {
+        return connectionConfigGet("krb5Conf", "");
+    }
+
+    default String getKrb5Principal() {
+        return connectionConfigGet("krb5Principal", "");
+    }
+
+    default String getKrb5ServiceName() {
+        return connectionConfigGet("krb5ServiceName", "");
+    }
+
     default String getSaslMechanism() {
         return connectionConfigGet(KEY_SASL_MECHANISM, "PLAIN");
     }