diff --git a/.run/Docker - Run Dev Stack - Kafka, Postgres, SEP24 Reference UI.run.xml b/.run/Docker - Run Dev Stack - Kafka, Postgres, SEP24 Reference UI.run.xml
index 2ea0db794..f5460fad0 100644
--- a/.run/Docker - Run Dev Stack - Kafka, Postgres, SEP24 Reference UI.run.xml	
+++ b/.run/Docker - Run Dev Stack - Kafka, Postgres, SEP24 Reference UI.run.xml	
@@ -1,9 +1,18 @@
 <component name="ProjectRunConfigurationManager">
   <configuration default="false" name="Docker - Run Dev Stack - Kafka, Postgres, SEP24 Reference UI" type="JetRunConfigurationType">
-    <option name="ALTERNATIVE_JRE_PATH_ENABLED" value="true" />
     <option name="MAIN_CLASS_NAME" value="org.stellar.anchor.platform.run_profiles.RunDockerDevStack" />
     <module name="java-stellar-anchor-sdk.service-runner.main" />
     <shortenClasspath name="ARGS_FILE" />
+    <extension name="net.ashald.envfile">
+      <option name="IS_ENABLED" value="false" />
+      <option name="IS_SUBST" value="false" />
+      <option name="IS_PATH_MACRO_SUPPORTED" value="false" />
+      <option name="IS_IGNORE_MISSING_FILES" value="false" />
+      <option name="IS_ENABLE_EXPERIMENTAL_INTEGRATIONS" value="false" />
+      <ENTRIES>
+        <ENTRY IS_ENABLED="true" PARSER="runconfig" IS_EXECUTABLE="false" />
+      </ENTRIES>
+    </extension>
     <method v="2">
       <option name="Make" enabled="true" />
     </method>
diff --git a/.run/Test Profile_ kafka-sasl-ssl.run.xml b/.run/Test Profile_ kafka-sasl-ssl.run.xml
new file mode 100644
index 000000000..8ec3d649d
--- /dev/null
+++ b/.run/Test Profile_ kafka-sasl-ssl.run.xml	
@@ -0,0 +1,25 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="Test Profile: kafka-sasl-ssl" type="JetRunConfigurationType" folderName="Run Test Profile">
+    <envs>
+      <env name="app.enableTest" value="true" />
+      <env name="KT_REFERENCE_SERVER_CONFIG" value="service-runner/src/main/resources/config/reference-config.yaml" />
+      <env name="TEST_PROFILE_NAME" value="kafka-sasl-ssl" />
+    </envs>
+    <option name="MAIN_CLASS_NAME" value="org.stellar.anchor.platform.run_profiles.RunTestProfile" />
+    <module name="java-stellar-anchor-sdk.service-runner.main" />
+    <shortenClasspath name="ARGS_FILE" />
+    <extension name="net.ashald.envfile">
+      <option name="IS_ENABLED" value="false" />
+      <option name="IS_SUBST" value="false" />
+      <option name="IS_PATH_MACRO_SUPPORTED" value="false" />
+      <option name="IS_IGNORE_MISSING_FILES" value="false" />
+      <option name="IS_ENABLE_EXPERIMENTAL_INTEGRATIONS" value="false" />
+      <ENTRIES>
+        <ENTRY IS_ENABLED="true" PARSER="runconfig" IS_EXECUTABLE="false" />
+      </ENTRIES>
+    </extension>
+    <method v="2">
+      <option name="Make" enabled="true" />
+    </method>
+  </configuration>
+</component>
\ No newline at end of file
diff --git a/platform/src/main/java/org/stellar/anchor/platform/config/KafkaConfig.java b/platform/src/main/java/org/stellar/anchor/platform/config/KafkaConfig.java
index abda4b69a..6b27c6341 100644
--- a/platform/src/main/java/org/stellar/anchor/platform/config/KafkaConfig.java
+++ b/platform/src/main/java/org/stellar/anchor/platform/config/KafkaConfig.java
@@ -32,18 +32,35 @@ public class KafkaConfig {
   /** Determines the maximum amount of time to wait for the batch to be filled. */
   int pollTimeoutSeconds;
 
+  /** The security protocol used to communicate with brokers. */
+  SecurityProtocol securityProtocol;
+
+  /** The SASL mechanism used for authentication. */
+  SaslMechanism saslMechanism;
+
+  /** the SSL keystore location. */
+  String sslKeystoreLocation;
+
+  /** the SSL truststore location. */
+  String sslTruststoreLocation;
+
   public enum SecurityProtocol {
     PLAINTEXT,
-    SASL_PLAINTEXT
+    SASL_PLAINTEXT,
+    SASL_SSL
   }
 
   public enum SaslMechanism {
-    PLAIN
-  }
+    PLAIN("PLAIN");
 
-  /** The security protocol used to communicate with brokers. */
-  SecurityProtocol securityProtocol;
+    String value;
 
-  /** The SASL mechanism used for authentication. */
-  SaslMechanism saslMechanism;
+    SaslMechanism(String value) {
+      this.value = value;
+    }
+
+    public String getValue() {
+      return value;
+    }
+  }
 }
diff --git a/platform/src/main/java/org/stellar/anchor/platform/config/MskConfig.java b/platform/src/main/java/org/stellar/anchor/platform/config/MskConfig.java
index 175da22f7..79f87e051 100644
--- a/platform/src/main/java/org/stellar/anchor/platform/config/MskConfig.java
+++ b/platform/src/main/java/org/stellar/anchor/platform/config/MskConfig.java
@@ -19,7 +19,10 @@ public class MskConfig extends KafkaConfig {
       int batchSize,
       int pollTimeoutSeconds,
       SecurityProtocol securityProtocol,
-      SaslMechanism saslMechanism) {
+      SaslMechanism saslMechanism,
+      String sslKeystoreLocation,
+      String sslTruststoreLocation) {
+
     super(
         bootstrapServer,
         clientId,
@@ -28,7 +31,9 @@ public class MskConfig extends KafkaConfig {
         batchSize,
         pollTimeoutSeconds,
         securityProtocol,
-        saslMechanism);
+        saslMechanism,
+        sslKeystoreLocation,
+        sslTruststoreLocation);
     this.useIAM = useIAM;
   }
 }
diff --git a/platform/src/main/java/org/stellar/anchor/platform/config/PropertySecretConfig.java b/platform/src/main/java/org/stellar/anchor/platform/config/PropertySecretConfig.java
index e832e10be..f61fb3dbb 100644
--- a/platform/src/main/java/org/stellar/anchor/platform/config/PropertySecretConfig.java
+++ b/platform/src/main/java/org/stellar/anchor/platform/config/PropertySecretConfig.java
@@ -21,6 +21,9 @@ public class PropertySecretConfig implements SecretConfig {
       "secret.events.queue.kafka.username";
   public static final String SECRET_EVENTS_QUEUE_KAFKA_PASSWORD =
       "secret.events.queue.kafka.password";
+  public static final String SECRET_SSL_KEYSTORE_PASSWORD = "secret.ssl.keystore.password";
+  public static final String SECRET_SSL_KEY_PASSWORD = "secret.ssl.key.password";
+  public static final String SECRET_SSL_TRUSTSTORE_PASSWORD = "secret.ssl.truststore.password";
 
   public String getSep6MoreInfoUrlJwtSecret() {
     return SecretManager.getInstance().get(SECRET_SEP_6_MORE_INFO_URL_JWT_SECRET);
diff --git a/platform/src/main/java/org/stellar/anchor/platform/configurator/SecretManager.java b/platform/src/main/java/org/stellar/anchor/platform/configurator/SecretManager.java
index d0cb429d5..32e8f315d 100644
--- a/platform/src/main/java/org/stellar/anchor/platform/configurator/SecretManager.java
+++ b/platform/src/main/java/org/stellar/anchor/platform/configurator/SecretManager.java
@@ -30,7 +30,10 @@ public class SecretManager
           PropertySecretConfig.SECRET_EVENTS_QUEUE_KAFKA_USERNAME,
           PropertySecretConfig.SECRET_EVENTS_QUEUE_KAFKA_PASSWORD,
           PropertyCustodySecretConfig.SECRET_FIREBLOCKS_SECRET_KEY,
-          PropertyCustodySecretConfig.SECRET_FIREBLOCKS_API_KEY);
+          PropertyCustodySecretConfig.SECRET_FIREBLOCKS_API_KEY,
+          PropertySecretConfig.SECRET_SSL_KEYSTORE_PASSWORD,
+          PropertySecretConfig.SECRET_SSL_KEY_PASSWORD,
+          PropertySecretConfig.SECRET_SSL_TRUSTSTORE_PASSWORD);
 
   final Properties props = new Properties();
 
@@ -42,6 +45,10 @@ public static SecretManager getInstance() {
     return secretManager;
   }
 
+  public static String secret(String key) {
+    return getInstance().get(key);
+  }
+
   @Override
   public void initialize(@NotNull ConfigurableApplicationContext applicationContext) {
     info("Secret manager started.");
diff --git a/platform/src/main/java/org/stellar/anchor/platform/event/DefaultEventService.java b/platform/src/main/java/org/stellar/anchor/platform/event/DefaultEventService.java
index 7ebb70708..605db5105 100644
--- a/platform/src/main/java/org/stellar/anchor/platform/event/DefaultEventService.java
+++ b/platform/src/main/java/org/stellar/anchor/platform/event/DefaultEventService.java
@@ -1,5 +1,6 @@
 package org.stellar.anchor.platform.event;
 
+import lombok.SneakyThrows;
 import org.apache.commons.lang3.NotImplementedException;
 import org.stellar.anchor.config.event.EventConfig;
 import org.stellar.anchor.event.EventService;
@@ -13,6 +14,7 @@ public DefaultEventService(EventConfig eventConfig) {
   }
 
   @Override
+  @SneakyThrows
   public Session createSession(String sessionName, EventQueue eventQueue) {
     if (eventConfig.isEnabled()) {
       switch (eventConfig.getQueue().getType()) {
diff --git a/platform/src/main/java/org/stellar/anchor/platform/event/KafkaSession.java b/platform/src/main/java/org/stellar/anchor/platform/event/KafkaSession.java
index 195c3c867..f2e1f305d 100644
--- a/platform/src/main/java/org/stellar/anchor/platform/event/KafkaSession.java
+++ b/platform/src/main/java/org/stellar/anchor/platform/event/KafkaSession.java
@@ -3,9 +3,16 @@
 import static org.apache.kafka.clients.CommonClientConfigs.SECURITY_PROTOCOL_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.*;
 import static org.apache.kafka.common.config.SaslConfigs.SASL_MECHANISM;
+import static org.apache.kafka.common.config.SslConfigs.*;
+import static org.stellar.anchor.platform.config.PropertySecretConfig.*;
+import static org.stellar.anchor.platform.configurator.SecretManager.*;
+import static org.stellar.anchor.platform.utils.ResourceHelper.findResourceFile;
+import static org.stellar.anchor.platform.utils.ResourceHelper.resource;
 import static org.stellar.anchor.util.StringHelper.isEmpty;
 
 import io.micrometer.core.instrument.Metrics;
+import java.io.File;
+import java.io.IOException;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
@@ -26,23 +33,39 @@
 import org.stellar.anchor.event.EventService;
 import org.stellar.anchor.event.EventService.EventQueue;
 import org.stellar.anchor.platform.config.KafkaConfig;
-import org.stellar.anchor.platform.config.PropertySecretConfig;
-import org.stellar.anchor.platform.configurator.SecretManager;
 import org.stellar.anchor.util.GsonUtils;
 import org.stellar.anchor.util.Log;
 
 public class KafkaSession implements EventService.Session {
-
   final KafkaConfig kafkaConfig;
   final String sessionName;
   final String topic;
   Producer<String, String> producer = null;
   Consumer<String, String> consumer = null;
+  String sslKeystoreLocation;
+  String sslTruststoreLocation;
 
-  KafkaSession(KafkaConfig kafkaConfig, String sessionName, EventQueue queue) {
+  KafkaSession(KafkaConfig kafkaConfig, String sessionName, EventQueue queue) throws IOException {
     this.kafkaConfig = kafkaConfig;
     this.sessionName = sessionName;
     this.topic = queue.name();
+
+    if (kafkaConfig.getSecurityProtocol() == KafkaConfig.SecurityProtocol.SASL_SSL) {
+      // If the keystore and truststore files exist, use them, otherwise, use the resources
+      if (new File(kafkaConfig.getSslKeystoreLocation()).exists()) {
+        sslKeystoreLocation = kafkaConfig.getSslKeystoreLocation();
+      } else {
+        sslKeystoreLocation =
+            findResourceFile(resource(kafkaConfig.getSslKeystoreLocation())).getAbsolutePath();
+      }
+
+      if (new File(kafkaConfig.getSslTruststoreLocation()).exists()) {
+        sslTruststoreLocation = kafkaConfig.getSslTruststoreLocation();
+      } else {
+        sslTruststoreLocation =
+            findResourceFile(resource(kafkaConfig.getSslTruststoreLocation())).getAbsolutePath();
+      }
+    }
   }
 
   @Override
@@ -174,37 +197,53 @@ Consumer<String, String> createConsumer() throws InvalidConfigException {
   void configureAuth(Properties props) throws InvalidConfigException {
     switch (kafkaConfig.getSecurityProtocol()) {
       case SASL_PLAINTEXT:
-        if (isEmpty(
-            SecretManager.getInstance()
-                .get(PropertySecretConfig.SECRET_EVENTS_QUEUE_KAFKA_USERNAME))) {
+      case SASL_SSL:
+        // Check if the username and password are set
+        if (isEmpty(secret(SECRET_EVENTS_QUEUE_KAFKA_USERNAME))) {
           String msg =
-              PropertySecretConfig.SECRET_EVENTS_QUEUE_KAFKA_USERNAME
+              SECRET_EVENTS_QUEUE_KAFKA_USERNAME
                   + " is not set. Please provide the Kafka username.";
           Log.error(msg);
           throw new InvalidConfigException(msg);
         }
-        if (isEmpty(
-            SecretManager.getInstance()
-                .get(PropertySecretConfig.SECRET_EVENTS_QUEUE_KAFKA_PASSWORD))) {
+        if (isEmpty(secret(SECRET_EVENTS_QUEUE_KAFKA_PASSWORD))) {
           String msg =
-              PropertySecretConfig.SECRET_EVENTS_QUEUE_KAFKA_PASSWORD
+              SECRET_EVENTS_QUEUE_KAFKA_PASSWORD
                   + " is not set. Please provide the Kafka password.";
           Log.error(msg);
           throw new InvalidConfigException(msg);
         }
 
-        props.put(SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
-        props.put(SASL_MECHANISM, "PLAIN");
+        // Set the SASL login information
         props.put(
             "sasl.jaas.config",
             "org.apache.kafka.common.security.plain.PlainLoginModule required username=\""
-                + SecretManager.getInstance()
-                    .get(PropertySecretConfig.SECRET_EVENTS_QUEUE_KAFKA_USERNAME)
+                + secret(SECRET_EVENTS_QUEUE_KAFKA_USERNAME)
                 + "\" password=\""
-                + SecretManager.getInstance()
-                    .get(PropertySecretConfig.SECRET_EVENTS_QUEUE_KAFKA_PASSWORD)
+                + secret(SECRET_EVENTS_QUEUE_KAFKA_PASSWORD)
                 + "\";");
+
         break;
+    }
+
+    switch (kafkaConfig.getSecurityProtocol()) {
+      case SASL_PLAINTEXT:
+        props.put(SECURITY_PROTOCOL_CONFIG, kafkaConfig.getSecurityProtocol().name());
+        props.put(SASL_MECHANISM, kafkaConfig.getSaslMechanism().getValue());
+        break;
+      case SASL_SSL:
+        props.put(SECURITY_PROTOCOL_CONFIG, kafkaConfig.getSecurityProtocol().name());
+        props.put(SASL_MECHANISM, kafkaConfig.getSaslMechanism().getValue());
+        props.put(SSL_KEYSTORE_LOCATION_CONFIG, sslKeystoreLocation);
+        props.put(SSL_TRUSTSTORE_LOCATION_CONFIG, sslTruststoreLocation);
+
+        if (!isEmpty(secret(SECRET_SSL_KEYSTORE_PASSWORD)))
+          props.put(SSL_KEYSTORE_PASSWORD_CONFIG, secret(SECRET_SSL_KEYSTORE_PASSWORD));
+        if (!isEmpty(secret(SECRET_SSL_KEY_PASSWORD)))
+          props.put(SSL_KEY_PASSWORD_CONFIG, secret(SECRET_SSL_KEY_PASSWORD));
+        if (!isEmpty(secret(SECRET_SSL_TRUSTSTORE_PASSWORD)))
+          props.put(SSL_TRUSTSTORE_PASSWORD_CONFIG, secret(SECRET_SSL_TRUSTSTORE_PASSWORD));
+
       case PLAINTEXT:
         break;
       default:
diff --git a/platform/src/main/java/org/stellar/anchor/platform/utils/ResourceHelper.java b/platform/src/main/java/org/stellar/anchor/platform/utils/ResourceHelper.java
new file mode 100644
index 000000000..890484b88
--- /dev/null
+++ b/platform/src/main/java/org/stellar/anchor/platform/utils/ResourceHelper.java
@@ -0,0 +1,22 @@
+package org.stellar.anchor.platform.utils;
+
+import java.io.File;
+import java.io.IOException;
+import org.springframework.core.io.DefaultResourceLoader;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.ResourceLoader;
+
+public class ResourceHelper {
+  static final ResourceLoader resourceLoader = new DefaultResourceLoader();
+
+  public static Resource resource(String resource) {
+    return resourceLoader.getResource(resource);
+  }
+
+  public static File findResourceFile(Resource resource) throws IOException {
+    if (resource.exists()) {
+      return resource.getFile();
+    }
+    throw new IOException("Resource not found: " + resource.getFilename());
+  }
+}
diff --git a/platform/src/main/resources/config/anchor-config-default-values.yaml b/platform/src/main/resources/config/anchor-config-default-values.yaml
index 973de0af8..62d3a78ce 100644
--- a/platform/src/main/resources/config/anchor-config-default-values.yaml
+++ b/platform/src/main/resources/config/anchor-config-default-values.yaml
@@ -681,15 +681,32 @@ events:
       poll_timeout_seconds: 60
       # The security protocol used to communicate with the brokers.
       # `security_protocol` can be one of the following:
-      #     `PLAINTEXT`: no encryption
-      #     `SASL_PLAINTEXT`: SASL authentication with no encryption
-      # If SASL_PLAINTEXT is used, the SECRET_EVENTS_QUEUE_KAFKA_USERNAME and SECRET_EVENTS_QUEUE_KAFKA_PASSWORD
-      # environment variables must be set.
+      #   `PLAINTEXT`: no encryption
+      #   `SASL_PLAINTEXT`: SASL authentication with no encryption
+      #   `SASL_SSL`: SASL authentication with SSL encryption
+      #
+      # If SASL_PLAINTEXT or SASL_SSL is used, the following secrets must be set
+      #   - SECRET_EVENTS_QUEUE_KAFKA_USERNAME
+      #   - SECRET_EVENTS_QUEUE_KAFKA_PASSWORD
+      #
+      # If the SASL_SSL is used, the following values must be set
+      #   - ssl_keystore_location
+      #   - ssl_keystore_password
+      # the following secrets must be set:
+      #   - SECRET_SSL_KEYSTORE_PASSWORD
+      #   - SECRET_SSL_TRUSTSTORE_PASSWORD
+      #   - SECRET_SSL_KEY_PASSWORD
+      #
       security_protocol: PLAINTEXT
-      # The SASL mechanism used for authentication when `security_protocol` is set to `SASL_PLAINTEXT`.
-      # `sasl_mechanism` can be one of the following:
-      #     `PLAIN`: PLAIN SASL mechanism
-      sasl_mechanism:
+      # The SASL mechanism used for authentication and must match what is used on the broker.
+      # `sasl_mechanism` can be one of the following supported mechanisms:
+      #   `PLAIN`: PLAIN SASL mechanism
+      sasl_mechanism: PLAIN
+      # The SSL keystore location. The file path to the keystore file (jks).
+      ssl_keystore_location:
+      # The SSL truststore location. The file path to the truststore file (jks).
+      ssl_truststore_location:
+
     # If the value of `publisher.type` is "sqs", the `events.publisher.sqs` field must be defined.
     sqs:
       # Use IAM authentication for AWS MSK or AWS SQS.
diff --git a/platform/src/main/resources/config/anchor-config-schema-v1.yaml b/platform/src/main/resources/config/anchor-config-schema-v1.yaml
index 0298df87c..72f0b8d62 100644
--- a/platform/src/main/resources/config/anchor-config-schema-v1.yaml
+++ b/platform/src/main/resources/config/anchor-config-schema-v1.yaml
@@ -35,6 +35,8 @@ events.queue.kafka.security_protocol:
 events.queue.kafka.sasl_mechanism:
 events.queue.kafka.username:
 events.queue.kafka.password:
+events.queue.kafka.ssl_keystore_location:
+events.queue.kafka.ssl_truststore_location:
 events.queue.msk.batch_size:
 events.queue.msk.bootstrap_server:
 events.queue.msk.client_id:
diff --git a/platform/src/test/kotlin/org/stellar/anchor/platform/config/EventConfigTest.kt b/platform/src/test/kotlin/org/stellar/anchor/platform/config/EventConfigTest.kt
index 66b5cfc29..680488be0 100644
--- a/platform/src/test/kotlin/org/stellar/anchor/platform/config/EventConfigTest.kt
+++ b/platform/src/test/kotlin/org/stellar/anchor/platform/config/EventConfigTest.kt
@@ -11,7 +11,6 @@ import org.springframework.validation.BindException
 import org.springframework.validation.Errors
 import org.springframework.validation.ValidationUtils
 import org.stellar.anchor.config.event.QueueConfig.QueueType.*
-import org.stellar.anchor.platform.config.*
 
 class EventConfigTest {
   lateinit var config: PropertyEventConfig
@@ -89,6 +88,8 @@ class EventConfigTest {
             10,
             KafkaConfig.SecurityProtocol.PLAINTEXT,
             null,
+            null,
+            null
           ),
         ),
         Arguments.of(
@@ -103,6 +104,8 @@ class EventConfigTest {
             10,
             KafkaConfig.SecurityProtocol.PLAINTEXT,
             null,
+            null,
+            null
           ),
         ),
         Arguments.of(
@@ -117,6 +120,8 @@ class EventConfigTest {
             10,
             KafkaConfig.SecurityProtocol.PLAINTEXT,
             null,
+            null,
+            null
           ),
         ),
         Arguments.of(
@@ -131,6 +136,8 @@ class EventConfigTest {
             10,
             KafkaConfig.SecurityProtocol.PLAINTEXT,
             null,
+            null,
+            null
           ),
         ),
         Arguments.of(
@@ -144,6 +151,8 @@ class EventConfigTest {
             500,
             10,
             KafkaConfig.SecurityProtocol.PLAINTEXT,
+            null,
+            null,
             null
           ),
         ),
@@ -159,12 +168,14 @@ class EventConfigTest {
             10,
             KafkaConfig.SecurityProtocol.PLAINTEXT,
             null,
+            null,
+            null
           ),
         ),
         Arguments.of(
           1,
           "kafka-security-protocol-empty",
-          KafkaConfig("localhost:29092", "client_id", 1, 10, 500, 10, null, null),
+          KafkaConfig("localhost:29092", "client_id", 1, 10, 500, 10, null, null, null, null),
         ),
         Arguments.of(
           1,
@@ -178,6 +189,8 @@ class EventConfigTest {
             10,
             KafkaConfig.SecurityProtocol.SASL_PLAINTEXT,
             null,
+            null,
+            null
           ),
         ),
       )
@@ -209,6 +222,8 @@ class EventConfigTest {
             10,
             KafkaConfig.SecurityProtocol.PLAINTEXT,
             null,
+            null,
+            null
           ),
         ),
         Arguments.of(
@@ -224,6 +239,8 @@ class EventConfigTest {
             10,
             KafkaConfig.SecurityProtocol.PLAINTEXT,
             null,
+            null,
+            null
           ),
         ),
         Arguments.of(
@@ -239,6 +256,8 @@ class EventConfigTest {
             10,
             KafkaConfig.SecurityProtocol.PLAINTEXT,
             null,
+            null,
+            null
           ),
         ),
         Arguments.of(
@@ -254,6 +273,8 @@ class EventConfigTest {
             10,
             KafkaConfig.SecurityProtocol.PLAINTEXT,
             null,
+            null,
+            null
           ),
         ),
         Arguments.of(
@@ -269,6 +290,8 @@ class EventConfigTest {
             10,
             KafkaConfig.SecurityProtocol.PLAINTEXT,
             null,
+            null,
+            null
           ),
         ),
         Arguments.of(
@@ -284,6 +307,8 @@ class EventConfigTest {
             10,
             KafkaConfig.SecurityProtocol.PLAINTEXT,
             null,
+            null,
+            null
           ),
         ),
       )
diff --git a/platform/src/test/kotlin/org/stellar/anchor/platform/utils/ResourceHelperTest.kt b/platform/src/test/kotlin/org/stellar/anchor/platform/utils/ResourceHelperTest.kt
new file mode 100644
index 000000000..d2b1413af
--- /dev/null
+++ b/platform/src/test/kotlin/org/stellar/anchor/platform/utils/ResourceHelperTest.kt
@@ -0,0 +1,17 @@
+package org.stellar.anchor.platform.utils
+
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
+
+class ResourceHelperTest {
+  @ParameterizedTest
+  @ValueSource(
+    strings = ["config/anchor-config-default-values.yaml", "config/anchor-config-schema-v1.yaml"]
+  )
+  fun `test reading resource file`(resource: String) {
+    val resource = ResourceHelper.resource(resource)
+    assertTrue(resource.exists())
+    assertTrue(resource.isFile)
+  }
+}
diff --git a/service-runner/src/main/kotlin/org/stellar/anchor/platform/TestHelper.kt b/service-runner/src/main/kotlin/org/stellar/anchor/platform/TestHelper.kt
index d65929cbc..9c29c5527 100644
--- a/service-runner/src/main/kotlin/org/stellar/anchor/platform/TestHelper.kt
+++ b/service-runner/src/main/kotlin/org/stellar/anchor/platform/TestHelper.kt
@@ -3,17 +3,16 @@ package org.stellar.anchor.platform
 import com.google.gson.Gson
 import java.io.InputStreamReader
 import java.nio.charset.StandardCharsets.UTF_8
-import org.springframework.core.io.DefaultResourceLoader
 import org.springframework.util.FileCopyUtils
+import org.stellar.anchor.platform.utils.ResourceHelper.resource
 import org.stellar.anchor.util.GsonUtils
 import org.stellar.anchor.util.Log.debug
 import org.stellar.anchor.util.StringHelper.json
 
 val gson: Gson = GsonUtils.getInstance()
-val resourceLoader = DefaultResourceLoader()
 
 fun resourceAsString(path: String): String {
-  val reader = InputStreamReader(resourceLoader.getResource(path).inputStream, UTF_8)
+  val reader = InputStreamReader(resource(path).inputStream, UTF_8)
   return FileCopyUtils.copyToString(reader)
 }
 
diff --git a/service-runner/src/main/resources/dev-tools/docker-compose-kafka-sasl.yaml b/service-runner/src/main/resources/dev-tools/docker-compose-kafka-sasl.yaml
new file mode 100644
index 000000000..42a6e8bae
--- /dev/null
+++ b/service-runner/src/main/resources/dev-tools/docker-compose-kafka-sasl.yaml
@@ -0,0 +1,48 @@
+version: '3'
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.4.3
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+    ports:
+      - "2181:2181"
+
+  kafka:
+    image: confluentinc/cp-kafka:7.4.3
+    depends_on:
+      - zookeeper
+    ports:
+      - "29092:29092"
+      - "29093:29093"
+      - "29094:29094"
+    environment:
+      # Zookeeper
+      ZOOKEEPER_SASL_ENABLED: "false"
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+
+      # General
+      KAFKA_BROKER_ID: 1
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
+
+      # Listeners
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "SSL:SSL,SASL_SSL:SASL_SSL,PLAINTEXT:PLAINTEXT"
+      KAFKA_LISTENERS: "SASL_SSL://kafka:29092,SSL://kafka:29093,PLAINTEXT://kafka:29094"
+      KAFKA_ADVERTISED_LISTENERS: "SASL_SSL://kafka:29092,SSL://kafka:29093,PLAINTEXT://kafka:29094"
+      KAFKA_INTER_BROKER_LISTENER_NAME: "PLAINTEXT"
+
+      # SASL
+      KAFKA_SASL_ENABLED_MECHANISMS: "PLAIN"
+      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: "PLAIN"
+      KAFKA_OPTS: "-Djava.security.auth.login.config=/etc/kafka/secrets/jaas.conf"
+
+      # SSL
+      KAFKA_SSL_KEYSTORE_FILENAME: kafka.keystore.jks
+      KAFKA_SSL_TRUSTSTORE_FILENAME: kafka.truststore.jks
+      KAFKA_SSL_KEYSTORE_CREDENTIALS: creds
+      KAFKA_SSL_KEY_CREDENTIALS: creds
+      KAFKA_SSL_TRUSTSTORE_CREDENTIALS: creds
+
+    volumes:
+      - ./secrets:/etc/kafka/secrets
\ No newline at end of file
diff --git a/service-runner/src/main/resources/dev-tools/secrets/creds b/service-runner/src/main/resources/dev-tools/secrets/creds
new file mode 100644
index 000000000..0c759a444
--- /dev/null
+++ b/service-runner/src/main/resources/dev-tools/secrets/creds
@@ -0,0 +1 @@
+test123
\ No newline at end of file
diff --git a/service-runner/src/main/resources/dev-tools/secrets/jaas.conf b/service-runner/src/main/resources/dev-tools/secrets/jaas.conf
new file mode 100644
index 000000000..697a95439
--- /dev/null
+++ b/service-runner/src/main/resources/dev-tools/secrets/jaas.conf
@@ -0,0 +1,9 @@
+KafkaServer {
+  org.apache.kafka.common.security.plain.PlainLoginModule required
+  username="admin"
+  password="admin-secret"
+  user_admin="admin-secret"
+  serviceName="kafka"
+  user_alice="alice-secret";
+};
+Client {};
\ No newline at end of file
diff --git a/service-runner/src/main/resources/dev-tools/secrets/kafka.crt b/service-runner/src/main/resources/dev-tools/secrets/kafka.crt
new file mode 100644
index 000000000..dbd168a7b
--- /dev/null
+++ b/service-runner/src/main/resources/dev-tools/secrets/kafka.crt
@@ -0,0 +1,8 @@
+-----BEGIN CERTIFICATE-----
+MIIBGjCBwQIUVvY9pKLNh+MrTEJmgX2pBAUQEm4wCgYIKoZIzj0EAwIwEDEOMAwG
+A1UEAwwFa2Fma2EwHhcNMjQwOTEyMTg1NTQwWhcNMjUwOTEyMTg1NTQwWjAQMQ4w
+DAYDVQQDDAVrYWZrYTBZMBMGByqGSM49AgEGCCqGSM49AwEHA0IABAGHI6vmIuKA
+gHARBUe6uQKmPT4Llj9x5HSFPLe94TZ9Rhn3YmUxccYOEUaEM/zg6Q97mDnwrt1V
+O1+9o8Muu00wCgYIKoZIzj0EAwIDSAAwRQIhAKtawIuPCmzZ6GaVWwxulH+cPDdH
+atVcTB8PNWfBVWCXAiAv9lbvdAXU4uz89Zutut8xrWYKLrMG2iv9EzDZpUY9mg==
+-----END CERTIFICATE-----
diff --git a/service-runner/src/main/resources/dev-tools/secrets/kafka.csr b/service-runner/src/main/resources/dev-tools/secrets/kafka.csr
new file mode 100644
index 000000000..d9d32f47f
--- /dev/null
+++ b/service-runner/src/main/resources/dev-tools/secrets/kafka.csr
@@ -0,0 +1,7 @@
+-----BEGIN CERTIFICATE REQUEST-----
+MIHKMHICAQAwEDEOMAwGA1UEAwwFa2Fma2EwWTATBgcqhkjOPQIBBggqhkjOPQMB
+BwNCAAQBhyOr5iLigIBwEQVHurkCpj0+C5Y/ceR0hTy3veE2fUYZ92JlMXHGDhFG
+hDP84OkPe5g58K7dVTtfvaPDLrtNoAAwCgYIKoZIzj0EAwIDSAAwRQIhAJa314Hc
+KiV8GjjWqJoS1Oz81YvnMGt7sVn5BHYlsEKoAiA0U+lmRkVu9LSIii6UR65E8+EW
+bR3HHFj9coo2BWGQOA==
+-----END CERTIFICATE REQUEST-----
diff --git a/service-runner/src/main/resources/dev-tools/secrets/kafka.key b/service-runner/src/main/resources/dev-tools/secrets/kafka.key
new file mode 100644
index 000000000..f61c987f8
--- /dev/null
+++ b/service-runner/src/main/resources/dev-tools/secrets/kafka.key
@@ -0,0 +1,5 @@
+-----BEGIN EC PRIVATE KEY-----
+MHcCAQEEIMHOlIpl6qlsf0529/Cui8J7ASo8Hi6Yd5PFlJ09DLTeoAoGCCqGSM49
+AwEHoUQDQgAEAYcjq+Yi4oCAcBEFR7q5AqY9PguWP3HkdIU8t73hNn1GGfdiZTFx
+xg4RRoQz/ODpD3uYOfCu3VU7X72jwy67TQ==
+-----END EC PRIVATE KEY-----
diff --git a/service-runner/src/main/resources/dev-tools/secrets/kafka.keystore.jks b/service-runner/src/main/resources/dev-tools/secrets/kafka.keystore.jks
new file mode 100644
index 000000000..f03d17de1
Binary files /dev/null and b/service-runner/src/main/resources/dev-tools/secrets/kafka.keystore.jks differ
diff --git a/service-runner/src/main/resources/dev-tools/secrets/kafka.p12 b/service-runner/src/main/resources/dev-tools/secrets/kafka.p12
new file mode 100644
index 000000000..5a53be68f
Binary files /dev/null and b/service-runner/src/main/resources/dev-tools/secrets/kafka.p12 differ
diff --git a/service-runner/src/main/resources/dev-tools/secrets/kafka.truststore.jks b/service-runner/src/main/resources/dev-tools/secrets/kafka.truststore.jks
new file mode 100644
index 000000000..746061963
Binary files /dev/null and b/service-runner/src/main/resources/dev-tools/secrets/kafka.truststore.jks differ
diff --git a/service-runner/src/main/resources/profiles/kafka-sasl-ssl/config.env b/service-runner/src/main/resources/profiles/kafka-sasl-ssl/config.env
new file mode 100644
index 000000000..ecaf32275
--- /dev/null
+++ b/service-runner/src/main/resources/profiles/kafka-sasl-ssl/config.env
@@ -0,0 +1,9 @@
+events.queue.kafka.security_protocol=SASL_SSL
+events.queue.kafka.ssl_keystore_location=classpath:dev-tools/secrets/kafka.keystore.jks
+events.queue.kafka.ssl_truststore_location=classpath:dev-tools/secrets/kafka.truststore.jks
+
+secret.events.queue.kafka.username=admin
+secret.events.queue.kafka.password=admin-secret
+secret.ssl.key.password=test123
+secret.ssl.keystore.password=test123
+secret.ssl.truststore.password=test123
diff --git a/service-runner/src/main/resources/profiles/kafka-sasl-ssl/test.env b/service-runner/src/main/resources/profiles/kafka-sasl-ssl/test.env
new file mode 100644
index 000000000..e69de29bb