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 8b19103f57..4cf340ae48 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 13ff573a12..f8d710384e 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)
@@ -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 68195c77f8..d981ed57f5 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 b8e43908ed..e748f567fc 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 3f6a64c627..b5c20cb3b5 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 081d4c44e1..e76cad50d0 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 0000000000..1fc439cf64
--- /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 730c9f916e..a52dffb362 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 600ee00c62..3f99df7d1c 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;
@@ -125,7 +124,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;
 
@@ -923,8 +921,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 6c3175ec65..ce343fe702 100644
--- a/python/hsfs/storage_connector.py
+++ b/python/hsfs/storage_connector.py
@@ -1124,6 +1124,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."""
@@ -1245,11 +1258,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
@@ -1266,9 +1285,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 4f14a3def3..6233e8d023 100644
--- a/python/tests/test_storage_connector.py
+++ b/python/tests/test_storage_connector.py
@@ -818,6 +818,108 @@ def test_confluent_options_jaas_double_quotes(self, mocker):
             "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:
     def test_from_response_json(self, backend_fixtures):