diff --git a/CHANGELOG.md b/CHANGELOG.md
index e430dd6d..4c8b3d4e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,10 @@
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
+## 2.6.0 (UNRELEASED)
+- [ISSUE-144](https://github.com/SourceLabOrg/kafka-webview/issues/144) Make providing a TrustStore file when setting up a SSL enabled cluster optional.  You might not want/need this option if your JVM is already configured to accept the SSL certificate served by the cluster, or if the cluster's certificate can be validated by a publically accessible CA.
+- [PR-215](https://github.com/SourceLabOrg/kafka-webview/pull/215) Improve errors displayed when using the `test cluster` functionality.
+
 ## 2.5.1 (05/19/2020)
 - [ISSUE-209](https://github.com/SourceLabOrg/kafka-webview/issues/209) Expose HealthCheck and App Info endpoints without requiring authentication.
   - Docker image now exposes port 9090 for Actuator end points.
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/configuration/cluster/ClusterConfigController.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/configuration/cluster/ClusterConfigController.java
index d8094cac..d9ec9b06 100644
--- a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/configuration/cluster/ClusterConfigController.java
+++ b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/configuration/cluster/ClusterConfigController.java
@@ -24,6 +24,7 @@
 
 package org.sourcelab.kafka.webview.ui.controller.configuration.cluster;
 
+import org.apache.kafka.common.errors.TimeoutException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.sourcelab.kafka.webview.ui.controller.BaseController;
@@ -136,6 +137,7 @@ public String editClusterForm(
         clusterForm.setSsl(cluster.isSslEnabled());
         clusterForm.setKeyStoreFilename(cluster.getKeyStoreFile());
         clusterForm.setTrustStoreFilename(cluster.getTrustStoreFile());
+        clusterForm.setUseTrustStore(clusterForm.hasTrustStoreFilename());
 
         // Set SASL options
         final SaslProperties saslProperties = saslUtility.decodeProperties(cluster);
@@ -181,11 +183,14 @@ public String clusterUpdate(
         if (clusterForm.getSsl()) {
             // If we're creating a new cluster
             if (!clusterForm.exists()) {
-                // Ensure that we have files uploaded
-                if (clusterForm.getTrustStoreFile() == null || clusterForm.getTrustStoreFile().isEmpty()) {
-                    bindingResult.addError(new FieldError(
-                        "clusterForm", "trustStoreFile", null, true, null, null, "Select a TrustStore JKS to upload")
-                    );
+                // Ensure that we have files uploaded for the truststore,
+                // but only if they elected to upload a truststore at all.
+                if (clusterForm.getUseTrustStore()) {
+                    if (clusterForm.getTrustStoreFile() == null || clusterForm.getTrustStoreFile().isEmpty()) {
+                        bindingResult.addError(new FieldError(
+                            "clusterForm", "trustStoreFile", null, true, null, null, "Select a TrustStore JKS to upload")
+                        );
+                    }
                 }
 
                 // Only require KeyStore if NOT using SASL
@@ -233,8 +238,23 @@ public String clusterUpdate(
             // Flip flag to true
             cluster.setSslEnabled(true);
 
-            // Determine if we should update keystores
-            if (!clusterForm.exists() || (clusterForm.getTrustStoreFile() != null && !clusterForm.getTrustStoreFile().isEmpty())) {
+            // If they've selected to NOT use a trust store.
+            if (!clusterForm.getUseTrustStore()) {
+                // Delete previous trust store if exists
+                if (cluster.getTrustStoreFile() != null) {
+                    uploadManager.deleteKeyStore(cluster.getTrustStoreFile());
+                }
+                // Clear out properties
+                cluster.setTrustStoreFile(null);
+                cluster.setTrustStorePassword(null);
+            }
+
+            /*
+             * Determine if we should update truststore.  We Update it in the following scenarios:
+             * - If the cluster is being newly created.
+             * - If they uploaded a trust store, and there previously was a truststore.
+             */
+            else if (!clusterForm.exists() || (clusterForm.getTrustStoreFile() != null && !clusterForm.getTrustStoreFile().isEmpty())) {
                 // Delete previous trust store if updating
                 if (cluster.getTrustStoreFile() != null) {
                     uploadManager.deleteKeyStore(cluster.getTrustStoreFile());
@@ -262,6 +282,7 @@ public String clusterUpdate(
                 }
             }
 
+            // Determine if we should update keystores
             if (!clusterForm.exists() || (clusterForm.getKeyStoreFile() != null && !clusterForm.getKeyStoreFile().isEmpty())) {
                 // Delete previous key store if updating, or if SASL is enabled.
                 if (clusterForm.getSasl() || cluster.getKeyStoreFile() != null) {
@@ -421,7 +442,10 @@ public String testCluster(@PathVariable final Long id, final RedirectAttributes
             }
         } catch (final Exception e) {
             // Collect all reasons.
-            final String reason = e.getMessage();
+            String reason = e.getMessage();
+            if (e instanceof TimeoutException) {
+                reason = reason + " (This may indicate an authentication or connection problem)";
+            }
 
             // Set error msg
             redirectAttributes.addFlashAttribute(
@@ -450,5 +474,12 @@ private void setupBreadCrumbs(final Model model, final String name, final String
         } else {
             manager.addCrumb("Clusters", null);
         }
+
+        // Add default trust store property.
+        String defaultTrustStore = System.getProperty("javax.net.ssl.trustStore", "<JRE_HOME>/lib/security/cacerts");
+        if (defaultTrustStore != null && defaultTrustStore.trim().isEmpty()) {
+            defaultTrustStore = "<JRE_HOME>/lib/security/cacerts";
+        }
+        model.addAttribute("defaultTrustStore", defaultTrustStore);
     }
 }
\ No newline at end of file
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/configuration/cluster/forms/ClusterForm.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/configuration/cluster/forms/ClusterForm.java
index c77094cf..b7e57276 100644
--- a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/configuration/cluster/forms/ClusterForm.java
+++ b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/configuration/cluster/forms/ClusterForm.java
@@ -45,6 +45,7 @@ public class ClusterForm {
 
     // SSL Options
     private Boolean ssl = false;
+    private Boolean useTrustStore = false;
 
     private MultipartFile trustStoreFile;
     private String trustStoreFilename;
@@ -141,6 +142,22 @@ public String getTrustStoreFilename() {
         }
     }
 
+    public Boolean getUseTrustStore() {
+        return useTrustStore;
+    }
+
+    public void setUseTrustStore(final Boolean useTrustStore) {
+        this.useTrustStore = useTrustStore;
+    }
+
+    /**
+     * Is there a configured TrustStore file?
+     * @return true if so, false if not.
+     */
+    public boolean hasTrustStoreFilename() {
+        return trustStoreFilename != null && !trustStoreFilename.isEmpty();
+    }
+
     public void setTrustStoreFilename(final String trustStoreFilename) {
         this.trustStoreFilename = trustStoreFilename;
     }
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaClientConfigUtil.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaClientConfigUtil.java
index fd45dd7b..2f3f4c9e 100644
--- a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaClientConfigUtil.java
+++ b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaClientConfigUtil.java
@@ -131,8 +131,11 @@ private void applySslSettings(final ClusterConfig clusterConfig, final Map<Strin
             config.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, keyStoreRootPath + "/" + clusterConfig.getKeyStoreFile());
             config.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, clusterConfig.getKeyStorePassword());
         }
-        config.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, keyStoreRootPath + "/" + clusterConfig.getTrustStoreFile());
-        config.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, clusterConfig.getTrustStorePassword());
+        // Only put Trust properties if one is defined
+        if (clusterConfig.getTrustStoreFile() != null) {
+            config.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, keyStoreRootPath + "/" + clusterConfig.getTrustStoreFile());
+            config.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, clusterConfig.getTrustStorePassword());
+        }
     }
 
     /**
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaOperations.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaOperations.java
index 34ea180e..4f36aa0d 100644
--- a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaOperations.java
+++ b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaOperations.java
@@ -600,11 +600,16 @@ private List<ConfigItem> describeResource(final ConfigResource configResource) {
         }
     }
 
-    private RuntimeException handleExecutionException(final ExecutionException e) {
-        if (e.getCause() != null && e.getCause() instanceof RuntimeException) {
-            return (RuntimeException) e.getCause();
+    /**
+     * Handle ExecutionException errors when raised.
+     * @param executionException The exception raised.
+     * @return Appropriate exception instance to throw.
+     */
+    private RuntimeException handleExecutionException(final ExecutionException executionException) {
+        if (executionException.getCause() != null && executionException.getCause() instanceof RuntimeException) {
+            return (RuntimeException) executionException.getCause();
         }
-        return new RuntimeException(e.getMessage(), e);
+        return new RuntimeException(executionException.getMessage(), executionException);
     }
 
     /**
diff --git a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/dto/ApiErrorResponse.java b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/dto/ApiErrorResponse.java
index f1ccb509..4602046e 100644
--- a/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/dto/ApiErrorResponse.java
+++ b/kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/dto/ApiErrorResponse.java
@@ -24,8 +24,6 @@
 
 package org.sourcelab.kafka.webview.ui.manager.kafka.dto;
 
-import net.bytebuddy.implementation.bytecode.Throw;
-
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
diff --git a/kafka-webview-ui/src/main/resources/templates/configuration/cluster/create.html b/kafka-webview-ui/src/main/resources/templates/configuration/cluster/create.html
index ea8e5520..694134cb 100644
--- a/kafka-webview-ui/src/main/resources/templates/configuration/cluster/create.html
+++ b/kafka-webview-ui/src/main/resources/templates/configuration/cluster/create.html
@@ -22,6 +22,10 @@
                 var isSaslChecked = jQuery('#sasl').is(':checked');
                 jQuery('#ssl-keystore-options').toggle(!isSaslChecked);
             });
+            jQuery('#useTrustStore').click(function() {
+                var isChecked = jQuery('#useTrustStore').is(':checked');
+                jQuery('#ssl-truststore-options').toggle(isChecked);
+            });
             jQuery('#sasl').click(function() {
                 var isChecked = jQuery('#sasl').is(':checked');
 
@@ -114,37 +118,64 @@ <h6>SSL Settings</h6>
 
                             <!-- SSL Options -->
                             <div id="ssl-options" th:styleappend="${clusterForm.getSsl()} ? 'display: block;' : 'display: none;'">
-                                <!-- Trust Store file -->
+                                <!-- Do you require upload of additional truststore? -->
                                 <div class="form-group row">
-                                    <label class="col-md-3 form-control-label" for="trustStoreFile">
-                                        Trust Store
-                                        <div
-                                            th:if="${clusterForm.getTrustStoreFilename()} != null"
-                                            th:text="'Current: ' + ${clusterForm.getTrustStoreFilename()}">
-                                        </div>
+                                    <label class="col-md-3 form-control-label" for="useTrustStore">
+                                        Upload truststore?
                                     </label>
                                     <div class="col-md-9">
                                         <input
-                                            id="trustStoreFile" name="trustStoreFile" class="form-control" type="file"
-                                            placeholder="Select TrustStore JKS"
+                                            id="useTrustStore" name="useTrustStore" class="" type="checkbox"
                                             th:errorclass="is-invalid"
-                                            th:value="*{trustStoreFile}">
-                                        <div class="invalid-feedback" th:if="${#fields.hasErrors('trustStoreFile')}" th:errors="*{trustStoreFile}"/>
+                                            th:field="*{useTrustStore}">
+
+                                        <small class="form-text text-muted">
+                                            Only required if the certificate served by your cluster is not registered with an accessible certificate authority (CA), or
+                                            if the cluster's certificate has not already been registered in your
+                                            JVM's default truststore<span th:if="${defaultTrustStore} == null">.</span>
+                                            <span th:if="${defaultTrustStore} != null"> at <i>[[${defaultTrustStore}]]</i></span>
+                                        </small>
                                     </div>
                                 </div>
 
-                                <!-- Trust Store Password -->
-                                <div class="form-group row">
-                                    <label class="col-md-3 form-control-label" for="trustStorePassword">
-                                        Trust Store Password
-                                    </label>
-                                    <div class="col-md-9">
-                                        <input
-                                            id="trustStorePassword" name="trustStorePassword" class="form-control" type="password"
-                                            placeholder="TrustStore password"
-                                            th:errorclass="is-invalid"
-                                            th:value="*{trustStorePassword}">
-                                        <div class="invalid-feedback" th:if="${#fields.hasErrors('trustStorePassword')}" th:errors="*{trustStorePassword}"/>
+                                <!-- Only display if trust store is configured -->
+                                <div id="ssl-truststore-options" th:styleappend="${clusterForm.hasTrustStoreFilename()} ? 'display: block;' : 'display: none;'">
+
+                                    <!-- Trust Store file -->
+                                    <div class="form-group row">
+                                        <label class="col-md-3 form-control-label" for="trustStoreFile">
+                                            Trust Store
+                                            <div
+                                                th:if="${clusterForm.getTrustStoreFilename()} != null"
+                                                th:text="'Current: ' + ${clusterForm.getTrustStoreFilename()}">
+                                            </div>
+                                        </label>
+                                        <div class="col-md-9">
+                                            <input
+                                                id="trustStoreFile" name="trustStoreFile" class="form-control" type="file"
+                                                placeholder="Select TrustStore JKS"
+                                                th:errorclass="is-invalid"
+                                                th:value="*{trustStoreFile}">
+                                            <div class="invalid-feedback" th:if="${#fields.hasErrors('trustStoreFile')}" th:errors="*{trustStoreFile}"/>
+                                            <small class="form-text text-muted" th:if="${clusterForm.getTrustStoreFilename()} != null">
+                                                Leave empty to continue using previously uploaded TrustStore.
+                                            </small>
+                                        </div>
+                                    </div>
+
+                                    <!-- Trust Store Password -->
+                                    <div class="form-group row">
+                                        <label class="col-md-3 form-control-label" for="trustStorePassword">
+                                            Trust Store Password
+                                        </label>
+                                        <div class="col-md-9">
+                                            <input
+                                                id="trustStorePassword" name="trustStorePassword" class="form-control" type="password"
+                                                placeholder="TrustStore password"
+                                                th:errorclass="is-invalid"
+                                                th:value="*{trustStorePassword}">
+                                            <div class="invalid-feedback" th:if="${#fields.hasErrors('trustStorePassword')}" th:errors="*{trustStorePassword}"/>
+                                        </div>
                                     </div>
                                 </div>
 
diff --git a/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/controller/configuration/cluster/ClusterConfigControllerTest.java b/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/controller/configuration/cluster/ClusterConfigControllerTest.java
index 0e2e7f87..972df768 100644
--- a/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/controller/configuration/cluster/ClusterConfigControllerTest.java
+++ b/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/controller/configuration/cluster/ClusterConfigControllerTest.java
@@ -208,11 +208,11 @@ public void testPostUpdate_existingNonSslCluster() throws Exception {
     }
 
     /**
-     * Test creating new ssl cluster.
+     * Test creating new ssl cluster with uploading a TrustStore.
      */
     @Test
     @Transactional
-    public void testPostUpdate_newSslCluster() throws Exception {
+    public void testPostUpdate_newSslCluster_withTrustStoreUploaded() throws Exception {
         final String expectedClusterName = "My New Cluster Name" + System.currentTimeMillis();
         final String expectedBrokerHosts = "localhost:9092";
         final String expectedKeystorePassword = "KeyStorePassword";
@@ -231,6 +231,7 @@ public void testPostUpdate_newSslCluster() throws Exception {
                 .param("name", expectedClusterName)
                 .param("brokerHosts", expectedBrokerHosts)
                 .param("ssl", "true")
+                .param("useTrustStore", "true")
                 .param("trustStorePassword", expectedTruststorePassword)
                 .param("keyStorePassword", expectedKeystorePassword)
             )
@@ -266,6 +267,59 @@ public void testPostUpdate_newSslCluster() throws Exception {
         Files.deleteIfExists(Paths.get(keyStoreUploadPath, cluster.getKeyStoreFile()));
     }
 
+    /**
+     * Test creating new ssl cluster WITHOUT uploading a TrustStore.
+     */
+    @Test
+    @Transactional
+    public void testPostUpdate_newSslCluster_withOutTrustStoreUploaded() throws Exception {
+        final String expectedClusterName = "My New Cluster Name" + System.currentTimeMillis();
+        final String expectedBrokerHosts = "localhost:9092";
+        final String expectedKeystorePassword = "KeyStorePassword";
+
+        final MockMultipartFile keyStoreUpload = new MockMultipartFile("keyStoreFile", "KeyStoreFile".getBytes(Charsets.UTF_8));
+
+        // Hit create page.
+        mockMvc
+            .perform(multipart("/configuration/cluster/update")
+                .file(keyStoreUpload)
+                .with(user(adminUserDetails))
+                .with(csrf())
+                .param("name", expectedClusterName)
+                .param("brokerHosts", expectedBrokerHosts)
+                .param("ssl", "true")
+                .param("keyStorePassword", expectedKeystorePassword)
+            )
+            //.andDo(print())
+            .andExpect(status().is3xxRedirection())
+            .andExpect(redirectedUrl("/configuration/cluster"));
+
+        // Lookup Cluster
+        final Cluster cluster = clusterRepository.findByName(expectedClusterName);
+        assertNotNull("Should have new cluster", cluster);
+        assertEquals("Has correct name", expectedClusterName, cluster.getName());
+        assertEquals("Has correct brokerHosts", expectedBrokerHosts, cluster.getBrokerHosts());
+        assertTrue("Should be ssl enabled", cluster.isSslEnabled());
+        assertFalse("Should not be valid by default", cluster.isValid());
+        assertNotNull("Should have passwords set", cluster.getKeyStorePassword());
+        assertNotNull("Should have key store file path", cluster.getKeyStoreFile());
+
+        // Should have null truststore file and password
+        assertNull("Should have trust store file path", cluster.getTrustStoreFile());
+        assertNull("Should have passwords set", cluster.getTrustStorePassword());
+
+        // NonSasl Validations
+        validateNonSaslCluster(cluster);
+
+        final boolean doesKeystoreFileExist = Files.exists(Paths.get(keyStoreUploadPath, cluster.getKeyStoreFile()));
+        assertTrue("KeyStore file should have been uploaded", doesKeystoreFileExist);
+        final String keyStoreContents = FileTestTools.readFile(keyStoreUploadPath + cluster.getKeyStoreFile());
+        assertEquals("KeyStore file should have correct contents", "KeyStoreFile", keyStoreContents);
+
+        // Cleanup
+        Files.deleteIfExists(Paths.get(keyStoreUploadPath, cluster.getKeyStoreFile()));
+    }
+
     /**
      * Test creating new ssl cluster but not providing a keystore will fail validation.
      */
@@ -316,12 +370,12 @@ public void testPostUpdate_newSslCluster_missingKeyStore_failsValidation() throw
     }
 
     /**
-     * Test updating existing ssl cluster, but not uploading new JKS files or updating passwords.
+     * Test updating existing ssl cluster with uploaded TrustStore, but not uploading new JKS files or updating passwords.
      * This should update the cluster, but leave the existing files and passwords in place.
      */
     @Test
     @Transactional
-    public void testPostUpdate_existingSslCluster() throws Exception {
+    public void testPostUpdate_existingSslCluster_previouslyExistingTrustStore() throws Exception {
         // Create an existing cluster
         final Cluster originalCluster = clusterTestTools.createCluster("My New Cluster");
         originalCluster.setValid(true);
@@ -349,6 +403,7 @@ public void testPostUpdate_existingSslCluster() throws Exception {
                 .param("name", expectedClusterName)
                 .param("brokerHosts", expectedBrokerHosts)
                 .param("ssl", "true")
+                .param("useTrustStore", "true")
             )
             //.andDo(print())
             .andExpect(status().is3xxRedirection())
@@ -384,8 +439,81 @@ public void testPostUpdate_existingSslCluster() throws Exception {
     }
 
     /**
-     * Test updating existing ssl cluster, updating the TrustStore file only.
-     * This should leave the keystore file/password as is.
+     * Test updating existing ssl cluster with uploaded TrustStore, but selecting to
+     * no longer use the uploaded trustStore.  In this case we should remove the truststore and clear the
+     * truststore password properties.
+     */
+    @Test
+    @Transactional
+    public void testPostUpdate_existingSslCluster_previouslyExistingTrustStore_butSelectToNoLongerUseIt() throws Exception {
+        // Define original truststore filename
+        final String originalTrustStoreFilename = "Cluster.TrustStore.jks";
+
+        // Create an existing cluster
+        final Cluster originalCluster = clusterTestTools.createCluster("My New Cluster");
+        originalCluster.setValid(true);
+        originalCluster.setSslEnabled(true);
+        originalCluster.setKeyStorePassword("DummyKeyStorePassword");
+        originalCluster.setKeyStoreFile("Cluster.KeyStore.jks");
+        originalCluster.setTrustStorePassword("DummyTrustStorePassword");
+        originalCluster.setTrustStoreFile(originalTrustStoreFilename);
+        clusterRepository.save(originalCluster);
+
+        // Create dummy JKS files
+        FileTestTools.createDummyFile(keyStoreUploadPath + originalCluster.getKeyStoreFile(), "KeyStoreFile");
+        FileTestTools.createDummyFile(keyStoreUploadPath + originalCluster.getTrustStoreFile(), "TrustStoreFile");
+
+        // Only update cluster name, brokers, keep SSL enabled.
+        final String expectedClusterName = "My Updated Cluster Name" + System.currentTimeMillis();
+        final String expectedBrokerHosts = "updatedHost:9092";
+
+        // Hit create page.
+        mockMvc
+            .perform(multipart("/configuration/cluster/update")
+                .with(user(adminUserDetails))
+                .with(csrf())
+                .param("id", String.valueOf(originalCluster.getId()))
+                .param("name", expectedClusterName)
+                .param("brokerHosts", expectedBrokerHosts)
+                .param("ssl", "true")
+            )
+            //.andDo(print())
+            .andExpect(status().is3xxRedirection())
+            .andExpect(redirectedUrl("/configuration/cluster"));
+
+        // Lookup Cluster
+        final Cluster cluster = clusterRepository.findById(originalCluster.getId()).get();
+        assertNotNull("Should have new cluster", cluster);
+        assertEquals("Has correct name", expectedClusterName, cluster.getName());
+        assertEquals("Has correct brokerHosts", expectedBrokerHosts, cluster.getBrokerHosts());
+        assertTrue("Should be ssl enabled", cluster.isSslEnabled());
+        assertFalse("Should not be valid by default", cluster.isValid());
+        assertEquals("Password should be unchanged", "DummyKeyStorePassword", cluster.getKeyStorePassword());
+        assertEquals("Should have key store file path", "Cluster.KeyStore.jks", cluster.getKeyStoreFile());
+
+        // Trust store should have been removed
+        assertNull("Should have trust store file path cleared", cluster.getTrustStoreFile());
+        assertNull("Password should have been cleared", cluster.getTrustStorePassword());
+
+        validateNonSaslCluster(cluster);
+
+        // Validate file exists
+        final boolean doesKeystoreFileExist = Files.exists(Paths.get(keyStoreUploadPath, cluster.getKeyStoreFile()));
+        assertTrue("KeyStore file should have been left untouched", doesKeystoreFileExist);
+        final String keyStoreContents = FileTestTools.readFile(keyStoreUploadPath + cluster.getKeyStoreFile());
+        assertEquals("KeyStore file should have remained untouched", "KeyStoreFile", keyStoreContents);
+
+        // Validate TrustStore file was removed from disk
+        final boolean doesTruststoreFileExist = Files.exists(Paths.get(keyStoreUploadPath, originalTrustStoreFilename));
+        assertFalse("trustStore file should have been removed from disk", doesTruststoreFileExist);
+
+        // Cleanup
+        Files.deleteIfExists(Paths.get(keyStoreUploadPath, cluster.getKeyStoreFile()));
+    }
+
+    /**
+     * Test updating existing ssl cluster with uploaded TrustStore, updating the TrustStore file only.
+     * This should leave the keystore file & keystore password as is.
      */
     @Test
     @Transactional
@@ -420,6 +548,7 @@ public void testPostUpdate_existingSslClusterUpdateTrustStore() throws Exception
                 .param("name", expectedClusterName)
                 .param("brokerHosts", expectedBrokerHosts)
                 .param("ssl", "true")
+                .param("useTrustStore", "true")
                 .param("trustStorePassword", "NewPassword")
             )
             //.andDo(print())
@@ -458,7 +587,7 @@ public void testPostUpdate_existingSslClusterUpdateTrustStore() throws Exception
     }
 
     /**
-     * Test updating existing ssl cluster, updating the KeyStore file only.
+     * Test updating existing ssl cluster with an uploaded TrustStore, updating the KeyStore file only.
      * This should leave the truststore file/password as is.
      */
     @Test
@@ -494,6 +623,7 @@ public void testPostUpdate_existingSslClusterUpdateKeyStore() throws Exception {
                 .param("name", expectedClusterName)
                 .param("brokerHosts", expectedBrokerHosts)
                 .param("ssl", "true")
+                .param("useTrustStore", "true")
                 .param("keyStorePassword", "NewPassword")
             )
             //.andDo(print())
@@ -799,13 +929,13 @@ public void testPostUpdate_existingSaslCluster() throws Exception {
     }
 
     /**
-     * Test creating new sasl PLAIN cluster with SSL.
+     * Test creating new sasl PLAIN cluster with SSL and uploaded TrustStore
      *
      * This should require a trust store, but no keystore or keystore password.
      */
     @Test
     @Transactional
-    public void testPostUpdate_newSaslSSL_Cluster() throws Exception {
+    public void testPostUpdate_newSaslSSL_Cluster_withUploadedTrustStore() throws Exception {
         final String expectedClusterName = "My New Cluster Name using SASL+SSL" + System.currentTimeMillis();
         final String expectedBrokerHosts = "localhost:9092";
         final String expectedSaslMechanism = "PLAIN";
@@ -825,6 +955,7 @@ public void testPostUpdate_newSaslSSL_Cluster() throws Exception {
                 .param("brokerHosts", expectedBrokerHosts)
                 // SSL Options
                 .param("ssl", "true")
+                .param("useTrustStore", "true")
                 .param("trustStorePassword", expectedTruststorePassword)
                 // SASL Options
                 .param("sasl", "true")
@@ -877,6 +1008,73 @@ public void testPostUpdate_newSaslSSL_Cluster() throws Exception {
         Files.deleteIfExists(Paths.get(keyStoreUploadPath, cluster.getTrustStoreFile()));
     }
 
+    /**
+     * Test creating new sasl PLAIN cluster with SSL and NO uploaded TrustStore
+     *
+     * This should not require a trust store, require no keystore or keystore password.
+     */
+    @Test
+    @Transactional
+    public void testPostUpdate_newSaslSSL_Cluster_withNoUploadedTrustStore() throws Exception {
+        final String expectedClusterName = "My New Cluster Name using SASL+SSL" + System.currentTimeMillis();
+        final String expectedBrokerHosts = "localhost:9092";
+        final String expectedSaslMechanism = "PLAIN";
+        final String expectedSaslUsername = "USERname";
+        final String expectedSaslPassword = "PASSword";
+
+        // Hit Update end point.
+        mockMvc
+            .perform(multipart("/configuration/cluster/update")
+                .with(user(adminUserDetails))
+                .with(csrf())
+                .param("name", expectedClusterName)
+                .param("brokerHosts", expectedBrokerHosts)
+                // SSL Options
+                .param("ssl", "true")
+                // SASL Options
+                .param("sasl", "true")
+                .param("saslMechanism", expectedSaslMechanism)
+                .param("saslUsername", expectedSaslUsername)
+                .param("saslPassword", expectedSaslPassword)
+            )
+            //.andDo(print())
+            .andExpect(status().is3xxRedirection())
+            .andExpect(redirectedUrl("/configuration/cluster"));
+
+        // Lookup Cluster
+        final Cluster cluster = clusterRepository.findByName(expectedClusterName);
+        assertNotNull("Should have new cluster", cluster);
+        assertEquals("Has correct name", expectedClusterName, cluster.getName());
+        assertEquals("Has correct brokerHosts", expectedBrokerHosts, cluster.getBrokerHosts());
+        assertTrue("Should be ssl enabled", cluster.isSslEnabled());
+        assertNull("Should NOT have keystore password set", cluster.getKeyStorePassword());
+        assertNull("Should NOT have key store file path", cluster.getKeyStoreFile());
+        assertFalse("Should not be valid by default", cluster.isValid());
+
+        // Should have no truststore properties set
+        assertNull("Should have no truststore password set", cluster.getTrustStorePassword());
+        assertNull("Should have no trust store file path", cluster.getTrustStoreFile());
+
+        // Validate sasl properties
+        assertTrue("Should have sasl enabled", cluster.isSaslEnabled());
+        assertEquals("Should have correct mechanism", expectedSaslMechanism, cluster.getSaslMechanism());
+        assertNotNull("Should have non-null sasl config", cluster.getSaslConfig());
+
+        // Attempt to decode encrypted config to validate
+        final SaslProperties saslProperties = saslUtility.decodeProperties(cluster);
+        assertNotNull("Should be non-null", saslProperties);
+        assertEquals("Should have expected mechanism", expectedSaslMechanism, saslProperties.getMechanism());
+        assertEquals("Should have expected username", expectedSaslUsername, saslProperties.getPlainUsername());
+        assertEquals("Should have expected password", expectedSaslPassword, saslProperties.getPlainPassword());
+        assertEquals(
+            "Should have default jaas for plain",
+            "org.apache.kafka.common.security.plain.PlainLoginModule required\n"
+                + "username=\"USERname\"\n"
+                + "password=\"PASSword\";",
+            saslProperties.getJaas()
+        );
+    }
+
     /**
      * Utility method for validating a cluster is not configured with SASL.
      * @param cluster cluster to validate.
diff --git a/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaClientConfigUtilTest.java b/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaClientConfigUtilTest.java
index a38dcf59..48b5b70b 100644
--- a/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaClientConfigUtilTest.java
+++ b/kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaClientConfigUtilTest.java
@@ -82,10 +82,10 @@ public void testApplyCommonSettings_noSsl_noSasl() {
     }
 
     /**
-     * Basic smoke test with SSL, without SASL options.
+     * Basic smoke test with SSL and TrustStore, without SASL options.
      */
     @Test
-    public void testApplyCommonSettings_withSsl_noSasl() {
+    public void testApplyCommonSettings_withSsl_noSasl_withTrustStore() {
         final ClusterConfig clusterConfig = ClusterConfig.newBuilder()
             .withBrokerHosts(expectedBrokerHosts)
             .withUseSsl(true)
@@ -100,7 +100,30 @@ public void testApplyCommonSettings_withSsl_noSasl() {
 
         // Validate
         validateDefaultKeys(config);
-        validateSsl(config, "SSL", true);
+        validateSsl(config, "SSL", true, true);
+        validateNoSasl(config);
+    }
+
+    /**
+     * Basic smoke test with SSL and NO TrustStore, without SASL options.
+     */
+    @Test
+    public void testApplyCommonSettings_withSsl_noSasl_withNoTrustStore() {
+        final ClusterConfig clusterConfig = ClusterConfig.newBuilder()
+            .withBrokerHosts(expectedBrokerHosts)
+            .withUseSsl(true)
+            .withKeyStoreFile(expectedKeyStoreFile)
+            .withKeyStorePassword(expectedKeyStorePassword)
+            .withTrustStoreFile(null)
+            .withTrustStorePassword(null)
+            .withUseSasl(false)
+            .build();
+
+        final Map<String, Object> config = util.applyCommonSettings(clusterConfig, consumerId);
+
+        // Validate
+        validateDefaultKeys(config);
+        validateSsl(config, "SSL", true, false);
         validateNoSasl(config);
     }
 
@@ -149,10 +172,10 @@ public void testApplyCommonSettings_noSsl_withSasl_plainMechanism() {
     }
 
     /**
-     * Basic smoke test with SSL, with SASL plain mechanism options.
+     * Basic smoke test with SSL and trust store, with SASL plain mechanism options.
      */
     @Test
-    public void testApplyCommonSettings_withSsl_withSasl() {
+    public void testApplyCommonSettings_withSsl_withSasl_withTrustStore() {
         final ClusterConfig clusterConfig = ClusterConfig.newBuilder()
             .withBrokerHosts(expectedBrokerHosts)
             .withUseSsl(true)
@@ -171,19 +194,52 @@ public void testApplyCommonSettings_withSsl_withSasl() {
 
         // Validate
         validateDefaultKeys(config);
-        validateSsl(config, "SASL_SSL", false);
+        validateSsl(config, "SASL_SSL", false, true);
+        validateSaslPlainMechanism(config, "SASL_SSL");
+    }
+
+    /**
+     * Basic smoke test with SSL and NO truststore, with SASL plain mechanism options.
+     */
+    @Test
+    public void testApplyCommonSettings_withSsl_withSasl_withNoTrustStore() {
+        final ClusterConfig clusterConfig = ClusterConfig.newBuilder()
+            .withBrokerHosts(expectedBrokerHosts)
+            .withUseSsl(true)
+            .withKeyStoreFile(expectedKeyStoreFile)
+            .withKeyStorePassword(expectedKeyStorePassword)
+            .withTrustStoreFile(null)
+            .withTrustStorePassword(null)
+            .withUseSasl(true)
+            .withSaslMechanism(expectedSaslPlainMechanism)
+            .withSaslPlaintextUsername(expectedSaslUsername)
+            .withSaslPlaintextPassword(expectedSaslPassword)
+            .withSaslJaas(expectedSaslJaas)
+            .build();
+
+        final Map<String, Object> config = util.applyCommonSettings(clusterConfig, consumerId);
+
+        // Validate
+        validateDefaultKeys(config);
+        validateSsl(config, "SASL_SSL", false, false);
         validateSaslPlainMechanism(config, "SASL_SSL");
     }
 
     private void validateSsl(
         final Map<String, Object> config,
         final String expectedSecurityProtocol,
-        final boolean shouldHaveKeyStoreConfiguration
+        final boolean shouldHaveKeyStoreConfiguration,
+        final boolean shouldHaveTrustStoreConfiguration
     ) {
         assertNotNull(config);
         validateKey(config, CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, expectedSecurityProtocol);
-        validateKey(config, SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "/tmp/" + expectedTrustStoreFile);
-        validateKey(config, SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, expectedTrustStorePassword);
+        if (shouldHaveTrustStoreConfiguration) {
+            validateKey(config, SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "/tmp/" + expectedTrustStoreFile);
+            validateKey(config, SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, expectedTrustStorePassword);
+        } else {
+            validateNoKey(config, SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG);
+            validateNoKey(config, SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG);
+        }
 
         if (shouldHaveKeyStoreConfiguration) {
             validateKey(config, SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, "/tmp/" + expectedKeyStoreFile);