From d8afcc1a4b79e8d2d9fe220adeccc7fb2c7a88cf Mon Sep 17 00:00:00 2001
From: bubriks <bubriks@gmail.com>
Date: Tue, 11 Feb 2025 14:36:02 +0200
Subject: [PATCH 1/3] init

---
 .../hsfs/beam/HopsworksConnection.java        |   3 +
 .../hsfs/beam/engine/BeamEngine.java          |   2 -
 .../hsfs/flink/HopsworksConnection.java       |   3 +
 .../hsfs/flink/engine/FlinkEngine.java        |   2 -
 .../logicalclocks/hsfs/StorageConnector.java  |  54 +++++++---
 .../logicalclocks/hsfs/engine/EngineBase.java |  10 ++
 .../hsfs/TestStorageConnector.java            |  65 +++++++++++
 .../hsfs/spark/HopsworksConnection.java       |   2 +
 .../hsfs/spark/engine/SparkEngine.java        |   5 +-
 python/hsfs/storage_connector.py              |  52 +++++++--
 python/tests/test_storage_connector.py        | 102 ++++++++++++++++++
 11 files changed, 273 insertions(+), 27 deletions(-)
 create mode 100644 java/hsfs/src/test/java/com/logicalclocks/hsfs/TestStorageConnector.java

diff --git a/java/beam/src/main/java/com/logicalclocks/hsfs/beam/HopsworksConnection.java b/java/beam/src/main/java/com/logicalclocks/hsfs/beam/HopsworksConnection.java
index 8b19103f5..4cf340ae4 100644
--- a/java/beam/src/main/java/com/logicalclocks/hsfs/beam/HopsworksConnection.java
+++ b/java/beam/src/main/java/com/logicalclocks/hsfs/beam/HopsworksConnection.java
@@ -20,6 +20,8 @@
 import com.logicalclocks.hsfs.FeatureStoreException;
 import com.logicalclocks.hsfs.HopsworksConnectionBase;
 import com.logicalclocks.hsfs.SecretStore;
+import com.logicalclocks.hsfs.beam.engine.BeamEngine;
+import com.logicalclocks.hsfs.engine.EngineBase;
 import com.logicalclocks.hsfs.metadata.Credentials;
 import com.logicalclocks.hsfs.metadata.HopsworksClient;
 import com.logicalclocks.hsfs.metadata.HopsworksHttpClient;
@@ -55,6 +57,7 @@ public HopsworksConnection(String host, int port, String project, Region region,
         hostnameVerification, trustStorePath, this.apiKeyFilePath, this.apiKeyValue);
     this.projectObj = getProject();
     HopsworksClient.getInstance().setProject(this.projectObj);
+    EngineBase.setInstance(BeamEngine.getInstance());
     Credentials credentials = HopsworksClient.getInstance().getCredentials();
     HopsworksHttpClient hopsworksHttpClient =  HopsworksClient.getInstance().getHopsworksHttpClient();
     hopsworksHttpClient.setTrustStorePath(credentials.gettStore());
diff --git a/java/beam/src/main/java/com/logicalclocks/hsfs/beam/engine/BeamEngine.java b/java/beam/src/main/java/com/logicalclocks/hsfs/beam/engine/BeamEngine.java
index 13ff573a1..63c68147f 100644
--- a/java/beam/src/main/java/com/logicalclocks/hsfs/beam/engine/BeamEngine.java
+++ b/java/beam/src/main/java/com/logicalclocks/hsfs/beam/engine/BeamEngine.java
@@ -87,8 +87,6 @@ public Map<String, String> getKafkaConfig(FeatureGroupBase featureGroup, Map<Str
 
     StorageConnector.KafkaConnector storageConnector =
         storageConnectorApi.getKafkaStorageConnector(featureGroup.getFeatureStore(), external);
-    storageConnector.setSslTruststoreLocation(addFile(storageConnector.getSslTruststoreLocation()));
-    storageConnector.setSslKeystoreLocation(addFile(storageConnector.getSslKeystoreLocation()));
 
     Map<String, String> config = storageConnector.kafkaOptions();
 
diff --git a/java/flink/src/main/java/com/logicalclocks/hsfs/flink/HopsworksConnection.java b/java/flink/src/main/java/com/logicalclocks/hsfs/flink/HopsworksConnection.java
index 68195c77f..d981ed57f 100644
--- a/java/flink/src/main/java/com/logicalclocks/hsfs/flink/HopsworksConnection.java
+++ b/java/flink/src/main/java/com/logicalclocks/hsfs/flink/HopsworksConnection.java
@@ -20,6 +20,8 @@
 import com.logicalclocks.hsfs.FeatureStoreException;
 import com.logicalclocks.hsfs.HopsworksConnectionBase;
 import com.logicalclocks.hsfs.SecretStore;
+import com.logicalclocks.hsfs.engine.EngineBase;
+import com.logicalclocks.hsfs.flink.engine.FlinkEngine;
 import com.logicalclocks.hsfs.metadata.Credentials;
 import com.logicalclocks.hsfs.metadata.HopsworksClient;
 
@@ -56,6 +58,7 @@ public HopsworksConnection(String host, int port, String project, Region region,
         hostnameVerification, trustStorePath, this.apiKeyFilePath, this.apiKeyValue);
     this.projectObj = getProject();
     HopsworksClient.getInstance().setProject(this.projectObj);
+    EngineBase.setInstance(FlinkEngine.getInstance());
     if (!System.getProperties().containsKey(HopsworksInternalClient.REST_ENDPOINT_SYS)) {
       Credentials credentials = HopsworksClient.getInstance().getCredentials();
       HopsworksHttpClient hopsworksHttpClient =  HopsworksClient.getInstance().getHopsworksHttpClient();
diff --git a/java/flink/src/main/java/com/logicalclocks/hsfs/flink/engine/FlinkEngine.java b/java/flink/src/main/java/com/logicalclocks/hsfs/flink/engine/FlinkEngine.java
index b8e43908e..e748f567f 100644
--- a/java/flink/src/main/java/com/logicalclocks/hsfs/flink/engine/FlinkEngine.java
+++ b/java/flink/src/main/java/com/logicalclocks/hsfs/flink/engine/FlinkEngine.java
@@ -128,8 +128,6 @@ public Map<String, String> getKafkaConfig(FeatureGroupBase featureGroup, Map<Str
 
     StorageConnector.KafkaConnector storageConnector =
             storageConnectorApi.getKafkaStorageConnector(featureGroup.getFeatureStore(), external);
-    storageConnector.setSslTruststoreLocation(addFile(storageConnector.getSslTruststoreLocation()));
-    storageConnector.setSslKeystoreLocation(addFile(storageConnector.getSslKeystoreLocation()));
 
     Map<String, String> config = storageConnector.kafkaOptions();
 
diff --git a/java/hsfs/src/main/java/com/logicalclocks/hsfs/StorageConnector.java b/java/hsfs/src/main/java/com/logicalclocks/hsfs/StorageConnector.java
index 3f6a64c62..b5c20cb3b 100644
--- a/java/hsfs/src/main/java/com/logicalclocks/hsfs/StorageConnector.java
+++ b/java/hsfs/src/main/java/com/logicalclocks/hsfs/StorageConnector.java
@@ -17,14 +17,27 @@
 
 package com.logicalclocks.hsfs;
 
+import java.io.IOException;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.google.common.base.Strings;
+import com.logicalclocks.hsfs.engine.EngineBase;
+import com.logicalclocks.hsfs.metadata.HopsworksClient;
 import com.logicalclocks.hsfs.metadata.HopsworksHttpClient;
 import com.logicalclocks.hsfs.metadata.Option;
 import com.logicalclocks.hsfs.metadata.StorageConnectorApi;
-import com.logicalclocks.hsfs.metadata.HopsworksClient;
 import com.logicalclocks.hsfs.util.Constants;
 
 import lombok.AllArgsConstructor;
@@ -32,17 +45,8 @@
 import lombok.NoArgsConstructor;
 import lombok.Setter;
 import lombok.ToString;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.utils.CollectionUtils;
 
-import java.io.IOException;
-import java.time.Instant;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
-
 @AllArgsConstructor
 @NoArgsConstructor
 @ToString
@@ -395,13 +399,13 @@ public static class KafkaConnector extends StorageConnector {
     @Getter @Setter
     protected SecurityProtocol securityProtocol;
 
-    @Getter @Setter
+    @Getter
     protected String sslTruststoreLocation;
 
     @Getter @Setter
     protected String sslTruststorePassword;
 
-    @Getter @Setter
+    @Getter
     protected String sslKeystoreLocation;
 
     @Getter @Setter
@@ -413,12 +417,36 @@ public static class KafkaConnector extends StorageConnector {
     @Getter @Setter
     protected SslEndpointIdentificationAlgorithm sslEndpointIdentificationAlgorithm;
 
-    @Getter @Setter
+    @Getter
     protected List<Option> options;
 
     @Getter @Setter
     protected Boolean externalKafka;
 
+    public void setSslTruststoreLocation(String sslTruststoreLocation) throws IOException, FeatureStoreException {
+      this.sslTruststoreLocation = EngineBase.getInstance().addFile(sslTruststoreLocation);
+    }
+
+    public void setSslKeystoreLocation(String sslKeystoreLocation) throws IOException, FeatureStoreException {
+      this.sslKeystoreLocation = EngineBase.getInstance().addFile(sslKeystoreLocation);
+    }
+
+    public void setOptions(List<Option> options) throws IOException, FeatureStoreException {
+      // add keytab file
+      for (Option option: options) {
+        if (option.getName().equals("sasl.jaas.config")) {
+          Pattern pattern = Pattern.compile("keyTab=[\"'](.+?)[\"']");
+          Matcher matcher = pattern.matcher(option.getValue());
+          while (matcher.find()) {
+            String originalKeytabLocation = matcher.group(1);
+            String newKeytabLocation = EngineBase.getInstance().addFile(originalKeytabLocation);
+            option.setValue(option.getValue().replace(originalKeytabLocation, newKeytabLocation));
+          }
+        }
+      }
+      this.options = options;
+    }
+
     public Map<String, String> kafkaOptions() throws FeatureStoreException {
       HopsworksHttpClient client = HopsworksClient.getInstance().getHopsworksHttpClient();
       Map<String, String> config = new HashMap<>();
diff --git a/java/hsfs/src/main/java/com/logicalclocks/hsfs/engine/EngineBase.java b/java/hsfs/src/main/java/com/logicalclocks/hsfs/engine/EngineBase.java
index 081d4c44e..e76cad50d 100644
--- a/java/hsfs/src/main/java/com/logicalclocks/hsfs/engine/EngineBase.java
+++ b/java/hsfs/src/main/java/com/logicalclocks/hsfs/engine/EngineBase.java
@@ -28,6 +28,16 @@
 
 public abstract class EngineBase {
 
+  private static EngineBase instance;
+
+  public static synchronized void setInstance(EngineBase instance) {
+    EngineBase.instance = instance;
+  }
+
+  public static synchronized EngineBase getInstance() {
+    return instance;
+  }
+
   protected static final Logger LOGGER = LoggerFactory.getLogger(EngineBase.class);
 
   public StorageConnectorApi storageConnectorApi = new StorageConnectorApi();
diff --git a/java/hsfs/src/test/java/com/logicalclocks/hsfs/TestStorageConnector.java b/java/hsfs/src/test/java/com/logicalclocks/hsfs/TestStorageConnector.java
new file mode 100644
index 000000000..1fc439cf6
--- /dev/null
+++ b/java/hsfs/src/test/java/com/logicalclocks/hsfs/TestStorageConnector.java
@@ -0,0 +1,65 @@
+/*
+ *  Copyright (c) 2025. Hopsworks AB
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *
+ *  See the License for the specific language governing permissions and limitations under the License.
+ *
+ */
+
+package com.logicalclocks.hsfs;
+
+import com.logicalclocks.hsfs.engine.EngineBase;
+import com.logicalclocks.hsfs.metadata.Option;
+
+import java.io.IOException;
+import java.util.Collections;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+ 
+ 
+class TestStorageConnector {
+  @Test
+  public void testOptions() throws FeatureStoreException, IOException {
+    // Arrange
+    EngineBase engineBase = Mockito.mock(EngineBase.class);
+    Mockito.when(engineBase.addFile(Mockito.anyString())).thenReturn("result_from_add_file");
+    EngineBase.setInstance(engineBase);
+
+    StorageConnector.KafkaConnector kafkaConnector = new StorageConnector.KafkaConnector();
+
+    // Act
+    kafkaConnector.setOptions(Collections.singletonList(new Option("test", "test")));
+
+    // Assert
+    Mockito.verify(engineBase, Mockito.times(0)).addFile(Mockito.anyString());
+  }
+
+  @Test
+  public void testOptionsSASLAuthentication() throws FeatureStoreException, IOException {
+    // Arrange
+    EngineBase engineBase = Mockito.mock(EngineBase.class);
+    Mockito.when(engineBase.addFile(Mockito.anyString())).thenReturn("result_from_add_file");
+    EngineBase.setInstance(engineBase);
+
+    StorageConnector.KafkaConnector kafkaConnector = new StorageConnector.KafkaConnector();
+
+    // Act
+    kafkaConnector.setOptions(Collections.singletonList(new Option("sasl.jaas.config", "com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab=\"/home/laurent/my.keytab\" storeKey=true useTicketCache=false serviceName=\"kafka\" principal=\"laurent@kafka.com\";")));
+
+    // Assert
+    Mockito.verify(engineBase, Mockito.times(1)).addFile(Mockito.anyString());
+    Assertions.assertEquals("com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab=\"result_from_add_file\" storeKey=true useTicketCache=false serviceName=\"kafka\" principal=\"laurent@kafka.com\";", kafkaConnector.getOptions().get(0).getValue());
+  }
+}
+ 
\ No newline at end of file
diff --git a/java/spark/src/main/java/com/logicalclocks/hsfs/spark/HopsworksConnection.java b/java/spark/src/main/java/com/logicalclocks/hsfs/spark/HopsworksConnection.java
index 730c9f916..a52dffb36 100644
--- a/java/spark/src/main/java/com/logicalclocks/hsfs/spark/HopsworksConnection.java
+++ b/java/spark/src/main/java/com/logicalclocks/hsfs/spark/HopsworksConnection.java
@@ -21,6 +21,7 @@
 import com.logicalclocks.hsfs.FeatureStoreException;
 import com.logicalclocks.hsfs.HopsworksConnectionBase;
 import com.logicalclocks.hsfs.SecretStore;
+import com.logicalclocks.hsfs.engine.EngineBase;
 import com.logicalclocks.hsfs.metadata.HopsworksClient;
 
 import com.logicalclocks.hsfs.metadata.HopsworksHttpClient;
@@ -53,6 +54,7 @@ public HopsworksConnection(String host, int port, String project, Region region,
         hostnameVerification, trustStorePath, this.apiKeyFilePath, this.apiKeyValue);
     this.projectObj = getProject();
     HopsworksClient.getInstance().setProject(this.projectObj);
+    EngineBase.setInstance(SparkEngine.getInstance());
     if (!System.getProperties().containsKey(HopsworksInternalClient.REST_ENDPOINT_SYS)) {
       SparkEngine.getInstance().validateSparkConfiguration();
       HopsworksHttpClient hopsworksHttpClient =  HopsworksClient.getInstance().getHopsworksHttpClient();
diff --git a/java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/SparkEngine.java b/java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/SparkEngine.java
index 600ee00c6..48614d962 100644
--- a/java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/SparkEngine.java
+++ b/java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/SparkEngine.java
@@ -125,7 +125,6 @@ public class SparkEngine extends EngineBase {
 
   private final StorageConnectorUtils storageConnectorUtils = new StorageConnectorUtils();
   private FeatureGroupUtils featureGroupUtils = new FeatureGroupUtils();
-  private OnlineIngestionApi onlineIngestionApi = new OnlineIngestionApi();
 
   private static SparkEngine INSTANCE = null;
 
@@ -144,7 +143,7 @@ public static void setInstance(SparkEngine sparkEngine) {
   @Getter
   private SparkSession sparkSession;
 
-  private SparkEngine() {
+  public SparkEngine() {
     sparkSession = SparkSession.builder()
         .enableHiveSupport()
         .getOrCreate();
@@ -923,8 +922,6 @@ public Map<String, String> getKafkaConfig(FeatureGroupBase featureGroup, Map<Str
 
     StorageConnector.KafkaConnector storageConnector =
         storageConnectorApi.getKafkaStorageConnector(featureGroup.getFeatureStore(), external);
-    storageConnector.setSslTruststoreLocation(addFile(storageConnector.getSslTruststoreLocation()));
-    storageConnector.setSslKeystoreLocation(addFile(storageConnector.getSslKeystoreLocation()));
 
     Map<String, String> config = storageConnector.sparkOptions();
 
diff --git a/python/hsfs/storage_connector.py b/python/hsfs/storage_connector.py
index 790e5d2c1..98d0b099f 100644
--- a/python/hsfs/storage_connector.py
+++ b/python/hsfs/storage_connector.py
@@ -1118,6 +1118,19 @@ def __init__(
         self._external_kafka = external_kafka
         self._pem_files_created = False
 
+        # add keytab file
+        sasl_jaas_config = self._options.get("sasl.jaas.config")
+        if sasl_jaas_config:
+            for option in re.findall("keyTab=[\"'](.+?)[\"']", sasl_jaas_config):
+                original_keytab_location = option
+                new_keytab_location = engine.get_instance().add_file(
+                    original_keytab_location
+                )
+                sasl_jaas_config = sasl_jaas_config.replace(
+                    original_keytab_location, new_keytab_location
+                )
+            self._options["sasl.jaas.config"] = sasl_jaas_config
+
     @property
     def bootstrap_servers(self) -> Optional[List[str]]:
         """Bootstrap servers string."""
@@ -1239,11 +1252,17 @@ def confluent_options(self) -> Dict[str, Any]:
                 config["ssl.key.location"] = client_key_path
             elif key == "sasl.jaas.config":
                 groups = re.search(
-                    "(.+?) .*username=[\"'](.+?)[\"'] .*password=[\"'](.+?)[\"']",
-                    value,
+                    "(.+) (required|requisite|sufficient|optional)(.*)", value
                 )
+                mechanism = groups.group(1)
+                # flag = groups.group(2)
+                options = groups.group(3)
+
+                option_dict = {}
+                for option in re.findall(r"\s(\w+)=[\"'](.+?)[\"']", options):
+                    option_dict[option[0]] = option[1]
+
                 if "sasl.mechanisms" not in config:
-                    mechanism = groups.group(1)
                     mechanism_value = None
                     if (
                         mechanism
@@ -1260,9 +1279,30 @@ def confluent_options(self) -> Dict[str, Any]:
                         == "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule"
                     ):
                         mechanism_value = "OAUTHBEARER"
-                config["sasl.mechanisms"] = mechanism_value
-                config["sasl.username"] = groups.group(2)
-                config["sasl.password"] = groups.group(3)
+                    else:
+                        mechanism_value = "GSSAPI"
+                    config["sasl.mechanisms"] = mechanism_value
+
+                    if mechanism_value == "GSSAPI":
+                        service_name = option_dict.get("serviceName")
+                        if service_name:
+                            config["sasl.kerberos.service.name"] = service_name
+
+                        principal = option_dict.get("principal")
+                        if principal:
+                            config["sasl.kerberos.principal"] = principal
+
+                        key_tab = option_dict.get("keyTab")
+                        if key_tab:
+                            config["sasl.kerberos.keytab"] = key_tab
+                    else:
+                        username = option_dict.get("username")
+                        if username:
+                            config["sasl.username"] = username
+
+                        password = option_dict.get("password")
+                        if password:
+                            config["sasl.password"] = password
             elif key == "ssl.endpoint.identification.algorithm":
                 config[key] = "none" if value == "" else value
             elif key == "queued.max.requests":
diff --git a/python/tests/test_storage_connector.py b/python/tests/test_storage_connector.py
index a1f7e77c3..ae24dd826 100644
--- a/python/tests/test_storage_connector.py
+++ b/python/tests/test_storage_connector.py
@@ -763,6 +763,108 @@ def test_confluent_options_jaas_double_quotes(self, mocker):
             "sasl.password": "111",
             "sasl.username": "222",
         }
+    
+    def test_kafka_options_kerberos(self, mocker, backend_fixtures):
+        # Arrange
+        mock_engine_get_instance = mocker.patch("hsfs.engine.get_instance")
+        mock_engine_get_instance.return_value.add_file.side_effect = [
+            None,
+            None,
+            "result_from_add_file",
+        ]
+        mocker.patch("hsfs.client.get_instance")
+        sc = storage_connector.KafkaConnector(
+            1,
+            "kafka_connector",
+            0,
+            external_kafka=True,
+            options=[
+                {
+                    "name": "sasl.jaas.config",
+                    "value": 'com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab="/home/laurent/my.keytab" storeKey=true useTicketCache=false serviceName="kafka" principal="laurent@kafka.com";',
+                }
+            ],
+        )
+
+        # Act
+        config = sc.kafka_options()
+
+        # Assert
+        assert config == {
+            "bootstrap.servers": None,
+            "security.protocol": None,
+            "ssl.endpoint.identification.algorithm": None,
+            "sasl.jaas.config": 'com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab="result_from_add_file" storeKey=true useTicketCache=false serviceName="kafka" principal="laurent@kafka.com";',
+        }
+
+    def test_spark_options_kerberos(self, mocker, backend_fixtures):
+        # Arrange
+        mock_engine_get_instance = mocker.patch("hsfs.engine.get_instance")
+        mock_engine_get_instance.return_value.add_file.side_effect = [
+            None,
+            None,
+            "result_from_add_file",
+        ]
+        mocker.patch("hsfs.client.get_instance")
+        sc = storage_connector.KafkaConnector(
+            1,
+            "kafka_connector",
+            0,
+            external_kafka=True,
+            options=[
+                {
+                    "name": "sasl.jaas.config",
+                    "value": 'com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab="/home/laurent/my.keytab" storeKey=true useTicketCache=false serviceName="kafka" principal="laurent@kafka.com";',
+                }
+            ],
+        )
+
+        # Act
+        config = sc.spark_options()
+
+        # Assert
+        assert config == {
+            "kafka.bootstrap.servers": None,
+            "kafka.security.protocol": None,
+            "kafka.ssl.endpoint.identification.algorithm": None,
+            "kafka.sasl.jaas.config": 'com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab="result_from_add_file" storeKey=true useTicketCache=false serviceName="kafka" principal="laurent@kafka.com";',
+        }
+
+    def test_confluent_options_kerberos(self, mocker):
+        # Arrange
+        mock_engine_get_instance = mocker.patch("hsfs.engine.get_instance")
+        mock_engine_get_instance.return_value.add_file.side_effect = [
+            None,
+            None,
+            "result_from_add_file",
+        ]
+        mocker.patch("hsfs.client.get_instance")
+        sc = storage_connector.KafkaConnector(
+            1,
+            "kafka_connector",
+            0,
+            external_kafka=True,
+            options=[
+                {
+                    "name": "sasl.jaas.config",
+                    "value": 'com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab="/home/laurent/my.keytab" storeKey=true useTicketCache=false serviceName="kafka" principal="laurent@kafka.com";',
+                }
+            ],
+        )
+
+        # Act
+        config = sc.confluent_options()
+
+        # Assert
+        assert config == {
+            "bootstrap.servers": None,
+            "security.protocol": None,
+            "ssl.endpoint.identification.algorithm": None,
+            "sasl.mechanisms": "GSSAPI",
+            "sasl.kerberos.service.name": "kafka",
+            "sasl.kerberos.principal": "laurent@kafka.com",
+            "sasl.kerberos.keytab": "result_from_add_file",
+        }
 
 
 class TestGcsConnector:

From 4aa4ebd8a07c47fbc95b346c5418572bdf556ffd Mon Sep 17 00:00:00 2001
From: bubriks <bubriks@gmail.com>
Date: Tue, 11 Feb 2025 15:12:47 +0200
Subject: [PATCH 2/3] some fix

---
 .../java/com/logicalclocks/hsfs/beam/engine/BeamEngine.java   | 4 ++--
 .../java/com/logicalclocks/hsfs/spark/engine/SparkEngine.java | 3 +--
 2 files changed, 3 insertions(+), 4 deletions(-)

diff --git a/java/beam/src/main/java/com/logicalclocks/hsfs/beam/engine/BeamEngine.java b/java/beam/src/main/java/com/logicalclocks/hsfs/beam/engine/BeamEngine.java
index 63c68147f..f8d710384 100644
--- a/java/beam/src/main/java/com/logicalclocks/hsfs/beam/engine/BeamEngine.java
+++ b/java/beam/src/main/java/com/logicalclocks/hsfs/beam/engine/BeamEngine.java
@@ -37,14 +37,14 @@ public class BeamEngine extends EngineBase {
   private static BeamEngine INSTANCE = null;
   private FeatureGroupUtils featureGroupUtils = new FeatureGroupUtils();
 
-  public static synchronized BeamEngine getInstance() throws FeatureStoreException {
+  public static synchronized BeamEngine getInstance() {
     if (INSTANCE == null) {
       INSTANCE = new BeamEngine();
     }
     return INSTANCE;
   }
 
-  private BeamEngine() throws FeatureStoreException {
+  private BeamEngine() {
   }
 
   public BeamProducer insertStream(StreamFeatureGroup streamFeatureGroup, Map<String, String> writeOptions)
diff --git a/java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/SparkEngine.java b/java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/SparkEngine.java
index 48614d962..3f99df7d1 100644
--- a/java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/SparkEngine.java
+++ b/java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/SparkEngine.java
@@ -41,7 +41,6 @@
 import com.logicalclocks.hsfs.FeatureGroupBase;
 import com.logicalclocks.hsfs.metadata.HopsworksClient;
 import com.logicalclocks.hsfs.metadata.OnDemandOptions;
-import com.logicalclocks.hsfs.metadata.OnlineIngestionApi;
 import com.logicalclocks.hsfs.metadata.Option;
 import com.logicalclocks.hsfs.util.Constants;
 import com.logicalclocks.hsfs.spark.ExternalFeatureGroup;
@@ -143,7 +142,7 @@ public static void setInstance(SparkEngine sparkEngine) {
   @Getter
   private SparkSession sparkSession;
 
-  public SparkEngine() {
+  private SparkEngine() {
     sparkSession = SparkSession.builder()
         .enableHiveSupport()
         .getOrCreate();

From 46c07e013667210be3bd8551bca469352c569bf0 Mon Sep 17 00:00:00 2001
From: bubriks <bubriks@gmail.com>
Date: Tue, 11 Feb 2025 15:15:06 +0200
Subject: [PATCH 3/3] ruff fix

---
 python/tests/test_storage_connector.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/python/tests/test_storage_connector.py b/python/tests/test_storage_connector.py
index ae24dd826..1ab79d8b8 100644
--- a/python/tests/test_storage_connector.py
+++ b/python/tests/test_storage_connector.py
@@ -763,7 +763,7 @@ def test_confluent_options_jaas_double_quotes(self, mocker):
             "sasl.password": "111",
             "sasl.username": "222",
         }
-    
+
     def test_kafka_options_kerberos(self, mocker, backend_fixtures):
         # Arrange
         mock_engine_get_instance = mocker.patch("hsfs.engine.get_instance")