diff --git a/core/pom.xml b/core/pom.xml
index 8c99423b1b..cecf765007 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -142,5 +142,10 @@
             <version>${guava.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>
diff --git a/core/src/main/java/io/confluent/rest/Application.java b/core/src/main/java/io/confluent/rest/Application.java
index 6bfbdd9521..b92318b163 100644
--- a/core/src/main/java/io/confluent/rest/Application.java
+++ b/core/src/main/java/io/confluent/rest/Application.java
@@ -545,7 +545,9 @@ public void join() throws InterruptedException {
    * @throws Exception If the application fails to stop
    */
   public void stop() throws Exception {
-    server.stop();
+    if (server != null) {
+      server.stop();
+    }
   }
 
   final void doShutdown() {
diff --git a/core/src/main/java/io/confluent/rest/ApplicationServer.java b/core/src/main/java/io/confluent/rest/ApplicationServer.java
index e031caed16..490acecd1a 100644
--- a/core/src/main/java/io/confluent/rest/ApplicationServer.java
+++ b/core/src/main/java/io/confluent/rest/ApplicationServer.java
@@ -54,6 +54,7 @@ public final class ApplicationServer<T extends RestConfig> extends Server {
   private final T config;
   private final ApplicationGroup applications;
   private final SslContextFactory sslContextFactory;
+  private FileWatcher sslKeystoreFileWatcher;
 
   private List<NetworkTrafficServerConnector> connectors = new ArrayList<>();
 
@@ -177,6 +178,9 @@ private void finalizeHandlerCollection(HandlerCollection handlers, HandlerCollec
   protected void doStop() throws Exception {
     super.doStop();
     applications.doStop();
+    if (sslKeystoreFileWatcher != null) {
+      sslKeystoreFileWatcher.shutdown();
+    }
   }
 
   protected final void doStart() throws Exception {
@@ -243,7 +247,9 @@ private Path getWatchLocation(RestConfig config) {
     return keystorePath;
   }
 
+  // CHECKSTYLE_RULES.OFF: CyclomaticComplexity|NPathComplexity
   private SslContextFactory createSslContextFactory(RestConfig config) {
+    // CHECKSTYLE_RULES.ON: CyclomaticComplexity|NPathComplexity
     SslContextFactory sslContextFactory = new SslContextFactory.Server();
     if (!config.getString(RestConfig.SSL_KEYSTORE_LOCATION_CONFIG).isEmpty()) {
       sslContextFactory.setKeyStorePath(
@@ -267,7 +273,9 @@ private SslContextFactory createSslContextFactory(RestConfig config) {
       if (config.getBoolean(RestConfig.SSL_KEYSTORE_RELOAD_CONFIG)) {
         Path watchLocation = getWatchLocation(config);
         try {
-          FileWatcher.onFileChange(watchLocation, () -> {
+          // create and shutdown a sslKeystoreFileWatcher for each Application, so that
+          //  all Applications in the same JVM don't use the same shared threadpool
+          sslKeystoreFileWatcher = FileWatcher.onFileChange(watchLocation, () -> {
                 // Need to reset the key store path for symbolic link case
                 sslContextFactory.setKeyStorePath(
                     config.getString(RestConfig.SSL_KEYSTORE_LOCATION_CONFIG)
@@ -301,9 +309,11 @@ private SslContextFactory createSslContextFactory(RestConfig config) {
       sslContextFactory.setTrustStorePath(
               config.getString(RestConfig.SSL_TRUSTSTORE_LOCATION_CONFIG)
       );
-      sslContextFactory.setTrustStorePassword(
-              config.getPassword(RestConfig.SSL_TRUSTSTORE_PASSWORD_CONFIG).value()
-      );
+      if (config.getPassword(RestConfig.SSL_TRUSTSTORE_PASSWORD_CONFIG) != null) {
+        sslContextFactory.setTrustStorePassword(
+            config.getPassword(RestConfig.SSL_TRUSTSTORE_PASSWORD_CONFIG).value()
+        );
+      }
       sslContextFactory.setTrustStoreType(
               config.getString(RestConfig.SSL_TRUSTSTORE_TYPE_CONFIG)
       );
diff --git a/core/src/main/java/io/confluent/rest/FileWatcher.java b/core/src/main/java/io/confluent/rest/FileWatcher.java
index c129847a95..e80652402e 100644
--- a/core/src/main/java/io/confluent/rest/FileWatcher.java
+++ b/core/src/main/java/io/confluent/rest/FileWatcher.java
@@ -35,7 +35,9 @@
 // reference https://gist.github.com/danielflower/f54c2fe42d32356301c68860a4ab21ed
 public class FileWatcher implements Runnable {
   private static final Logger log = LoggerFactory.getLogger(FileWatcher.class);
-  private static final ExecutorService executor = Executors.newFixedThreadPool(1,
+
+  // don't have static shared threadpool for all FileWatchers in the JVM
+  private final ExecutorService executor = Executors.newFixedThreadPool(1,
         new ThreadFactory() {
           public Thread newThread(Runnable r) {
             Thread t = Executors.defaultThreadFactory().newThread(r);
@@ -67,10 +69,11 @@ public FileWatcher(Path file, Callback callback) throws IOException {
     * Starts watching a file calls the callback when it is changed.
     * A shutdown hook is registered to stop watching.
   */
-  public static void onFileChange(Path file, Callback callback) throws IOException {
+  public static FileWatcher onFileChange(Path file, Callback callback) throws IOException {
     log.info("Configure watch file change: " + file);
     FileWatcher fileWatcher = new FileWatcher(file, callback);
-    executor.submit(fileWatcher);
+    fileWatcher.executor.submit(fileWatcher);
+    return fileWatcher;
   }
 
   public void run() {
diff --git a/core/src/main/java/io/confluent/rest/RestConfig.java b/core/src/main/java/io/confluent/rest/RestConfig.java
index 54fba9c2e1..ce0a4acece 100644
--- a/core/src/main/java/io/confluent/rest/RestConfig.java
+++ b/core/src/main/java/io/confluent/rest/RestConfig.java
@@ -161,7 +161,7 @@ public class RestConfig extends AbstractConfig {
   public static final String SSL_TRUSTSTORE_PASSWORD_CONFIG = "ssl.truststore.password";
   protected static final String SSL_TRUSTSTORE_PASSWORD_DOC =
       "The store password for the trust store file.";
-  protected static final String SSL_TRUSTSTORE_PASSWORD_DEFAULT = "";
+  protected static final String SSL_TRUSTSTORE_PASSWORD_DEFAULT = null;
   public static final String SSL_TRUSTSTORE_TYPE_CONFIG = "ssl.truststore.type";
   protected static final String SSL_TRUSTSTORE_TYPE_DOC =
       "The type of trust store file.";
diff --git a/core/src/test/java/io/confluent/rest/ApiHeadersTest.java b/core/src/test/java/io/confluent/rest/ApiHeadersTest.java
index ef5a0e13eb..bf24d7345b 100644
--- a/core/src/test/java/io/confluent/rest/ApiHeadersTest.java
+++ b/core/src/test/java/io/confluent/rest/ApiHeadersTest.java
@@ -45,6 +45,7 @@
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
 
 public class ApiHeadersTest {
 
@@ -56,11 +57,16 @@ public class ApiHeadersTest {
   private static String clientKeystoreLocation;
   private static TestApplication app;
 
+  // Use a temporary folder so that .jks files created by this test are isolated
+  //  and deleted when the test is done
+  public static TemporaryFolder tempFolder = new TemporaryFolder();
+
   @BeforeClass
   public static void setUp() throws Exception {
-    final File trustStore = File.createTempFile("ApiHeadersTest-truststore", ".jks");
-    final File clientKeystore = File.createTempFile("ApiHeadersTest-client-keystore", ".jks");
-    final File serverKeystore = File.createTempFile("ApiHeadersTest-server-keystore", ".jks");
+    tempFolder.create();
+    final File trustStore = File.createTempFile("ApiHeadersTest-truststore", ".jks", tempFolder.getRoot());
+    final File clientKeystore = File.createTempFile("ApiHeadersTest-client-keystore", ".jks", tempFolder.getRoot());
+    final File serverKeystore = File.createTempFile("ApiHeadersTest-server-keystore", ".jks", tempFolder.getRoot());
 
     clientKeystoreLocation = clientKeystore.getAbsolutePath();
 
@@ -84,6 +90,7 @@ public static void teardown() throws Exception {
     if (app != null) {
       app.stop();
     }
+    tempFolder.delete();
   }
 
   @Test
@@ -130,7 +137,7 @@ private static void createKeystoreWithCert(File file, String alias, Map<String,
     final X509Certificate cert = new CertificateBuilder(30, "SHA1withRSA")
         .sanDnsName("localhost").generate("CN=mymachine.local, O=A client", keypair);
 
-    TestSslUtils.createKeyStore(file.getPath(), new Password(SSL_PASSWORD), alias,
+    TestSslUtils.createKeyStore(file.getPath(), new Password(SSL_PASSWORD), new Password(SSL_PASSWORD), alias,
         keypair.getPrivate(), cert);
     certs.put(alias, cert);
   }
diff --git a/core/src/test/java/io/confluent/rest/SslTest.java b/core/src/test/java/io/confluent/rest/SslTest.java
index 3c70a4a2ea..64f40bccc6 100644
--- a/core/src/test/java/io/confluent/rest/SslTest.java
+++ b/core/src/test/java/io/confluent/rest/SslTest.java
@@ -18,6 +18,7 @@
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 
+import java.util.concurrent.TimeUnit;
 import org.apache.http.client.ClientProtocolException;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpGet;
@@ -30,8 +31,10 @@
 import org.apache.kafka.common.config.types.Password;
 import org.apache.kafka.test.TestSslUtils;
 import org.apache.kafka.test.TestSslUtils.CertificateBuilder;
-import org.junit.Before;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
 import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -49,7 +52,6 @@
 
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLException;
-import javax.net.ssl.SSLHandshakeException;
 import javax.ws.rs.GET;
 import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
@@ -61,28 +63,49 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertTrue;
+import static org.awaitility.Awaitility.await;
+import static org.junit.Assert.fail;
 
 public class SslTest {
   private static final Logger log = LoggerFactory.getLogger(SslTest.class);
 
-  private File trustStore;
-  private File clientKeystore;
-  private File serverKeystore;
-  private File serverKeystoreBak;
-  private File serverKeystoreErr;
+  private static File trustStore;
+  private static File clientKeystore;
+  private static File serverKeystore;
+  private static File serverKeystoreBak;
+  private static File serverKeystoreErr;
 
   public static final String SSL_PASSWORD = "test1234";
   public static final String EXPECTED_200_MSG = "Response status must be 200.";
   public static final int CERT_RELOAD_WAIT_TIME = 20000;
 
-  @Before
-  public void setUp() throws Exception {
+  private static TemporaryFolder tempFolder;
+
+  @BeforeClass
+  public static void setUp() throws Exception {
+
+    /*
+     * To make this test less flakey
+     *  - 1 don't create keystore files for every test method
+     *  - 2 cleanup keystore files on test exist so they don't have to be considerd by the FileWatcher
+     *  - 3 updated the FileWatcher and Application class to not have a single shared threadpool to
+     *      watch for changed files.
+     *
+     * By default temp files are not cleaned when up.  Which isn't normally a problem unless you are
+     * testing the ability of rest-utils apps to notice and reload updated ssl keystore files.
+     *
+     * Turns out the temp dir that Java+MacOs was continually using on my local machine had 1500
+     * files in it.  Also the Java "FileWatcher" for Mac works via polling a directory,
+     * this seems to have added to the flakeyness.
+     */
+    tempFolder = new TemporaryFolder();
+    tempFolder.create();
     try {
-      trustStore = File.createTempFile("SslTest-truststore", ".jks");
-      clientKeystore = File.createTempFile("SslTest-client-keystore", ".jks");
-      serverKeystore = File.createTempFile("SslTest-server-keystore", ".jks");
-      serverKeystoreBak = File.createTempFile("SslTest-server-keystore", ".jks.bak");
-      serverKeystoreErr = File.createTempFile("SslTest-server-keystore", ".jks.err");
+      trustStore = File.createTempFile("SslTest-truststore", ".jks", tempFolder.getRoot());
+      clientKeystore = File.createTempFile("SslTest-client-keystore", ".jks", tempFolder.getRoot());
+      serverKeystore = File.createTempFile("SslTest-server-keystore", ".jks", tempFolder.getRoot());
+      serverKeystoreBak = File.createTempFile("SslTest-server-keystore", ".jks.bak", tempFolder.getRoot());
+      serverKeystoreErr = File.createTempFile("SslTest-server-keystore", ".jks.err", tempFolder.getRoot());
     } catch (IOException ioe) {
       throw new RuntimeException("Unable to create temporary files for trust stores and keystores.");
     }
@@ -96,12 +119,17 @@ public void setUp() throws Exception {
     createWrongKeystoreWithCert(serverKeystoreErr, "server", certs);
   }
 
-  private void createKeystoreWithCert(File file, String alias, Map<String, X509Certificate> certs) throws Exception {
+  @AfterClass
+  public static void teardown() {
+    tempFolder.delete();
+  }
+
+  private static void createKeystoreWithCert(File file, String alias, Map<String, X509Certificate> certs) throws Exception {
     KeyPair keypair = TestSslUtils.generateKeyPair("RSA");
     CertificateBuilder certificateBuilder = new CertificateBuilder(30, "SHA1withRSA");
     X509Certificate cCert = certificateBuilder.sanDnsName("localhost")
         .generate("CN=mymachine.local, O=A client", keypair);
-    TestSslUtils.createKeyStore(file.getPath(), new Password(SSL_PASSWORD), alias, keypair.getPrivate(), cCert);
+    TestSslUtils.createKeyStore(file.getPath(), new Password(SSL_PASSWORD), new Password(SSL_PASSWORD), alias, keypair.getPrivate(), cCert);
     certs.put(alias, cCert);
   }
 
@@ -116,16 +144,25 @@ private void configServerTruststore(Properties props) {
     props.put(RestConfig.SSL_TRUSTSTORE_PASSWORD_CONFIG, SSL_PASSWORD);
   }
 
+  private void configServerTruststore(Properties props, String password) {
+    props.put(RestConfig.SSL_TRUSTSTORE_LOCATION_CONFIG, trustStore.getAbsolutePath());
+    props.put(RestConfig.SSL_TRUSTSTORE_PASSWORD_CONFIG, password);
+  }
+
+  private void configServerNoTruststorePassword(Properties props) {
+    props.put(RestConfig.SSL_TRUSTSTORE_LOCATION_CONFIG, trustStore.getAbsolutePath());
+  }
+
   private void enableSslClientAuth(Properties props) {
     props.put(RestConfig.SSL_CLIENT_AUTH_CONFIG, true);
   }
 
-  private void createWrongKeystoreWithCert(File file, String alias, Map<String, X509Certificate> certs) throws Exception {
+  private static void createWrongKeystoreWithCert(File file, String alias, Map<String, X509Certificate> certs) throws Exception {
     KeyPair keypair = TestSslUtils.generateKeyPair("RSA");
     CertificateBuilder certificateBuilder = new CertificateBuilder(30, "SHA1withRSA");
     X509Certificate cCert = certificateBuilder.sanDnsName("fail")
         .generate("CN=mymachine.local, O=A client", keypair);
-    TestSslUtils.createKeyStore(file.getPath(), new Password(SSL_PASSWORD), alias, keypair.getPrivate(), cCert);
+    TestSslUtils.createKeyStore(file.getPath(), new Password(SSL_PASSWORD), new Password(SSL_PASSWORD), alias, keypair.getPrivate(), cCert);
     certs.put(alias, cCert);
   }
 
@@ -168,30 +205,44 @@ public void testHttpsWithAutoReload() throws Exception {
     SslTestApplication app = new SslTestApplication(config);
     try {
       app.start();
-      int statusCode = makeGetRequest(httpsUri + "/test",
+      int startingCode = makeGetRequest(httpsUri + "/test",
                                   clientKeystore.getAbsolutePath(), SSL_PASSWORD, SSL_PASSWORD);
-      assertEquals(EXPECTED_200_MSG, 200, statusCode);
+      assertEquals(EXPECTED_200_MSG, 200, startingCode);
       assertMetricsCollected();
 
       // verify reload -- override the server keystore with a wrong one
       Files.copy(serverKeystoreErr.toPath(), serverKeystore.toPath(), StandardCopyOption.REPLACE_EXISTING);
-      Thread.sleep(CERT_RELOAD_WAIT_TIME);
-      boolean hitError = false;
-      try {
-        makeGetRequest(httpsUri + "/test",
-                                  clientKeystore.getAbsolutePath(), SSL_PASSWORD, SSL_PASSWORD);
-      } catch (Exception e) {
-        System.out.println(e);
-        hitError = true;
-      }
+      log.info("\tKeystore reload test : Applied bad keystore file");
+
+      await().pollInterval(2, TimeUnit.SECONDS).atMost(30, TimeUnit.SECONDS).untilAsserted( () -> {
+        boolean hitError = false;
+        try {
+          log.info("\tKeystore reload test : Awaiting failed https connection");
+          makeGetRequest(httpsUri + "/test", clientKeystore.getAbsolutePath(), SSL_PASSWORD, SSL_PASSWORD);
+        } catch (Exception e) {
+          System.out.println(e);
+          hitError = true;
+        }
+        assertTrue("Expecting to hit an error with new server cert", hitError);
+      });
 
       // verify reload -- override the server keystore with a correct one
       Files.copy(serverKeystoreBak.toPath(), serverKeystore.toPath(), StandardCopyOption.REPLACE_EXISTING);
-      Thread.sleep(CERT_RELOAD_WAIT_TIME);
-      statusCode = makeGetRequest(httpsUri + "/test",
-                                  clientKeystore.getAbsolutePath(), SSL_PASSWORD, SSL_PASSWORD);
-      assertEquals(EXPECTED_200_MSG, 200, statusCode); 
-      assertEquals("expect hit error with new server cert", true, hitError); 
+      log.info("\tKeystore reload test : keystore set back to good value");
+
+      await().pollInterval(2, TimeUnit.SECONDS).atMost(30, TimeUnit.SECONDS).untilAsserted( () -> {
+        try {
+          log.info("\tKeystore reload test : Awaiting a valid https connection");
+          int statusCode = makeGetRequest(httpsUri + "/test", clientKeystore.getAbsolutePath(), SSL_PASSWORD, SSL_PASSWORD);
+          assertEquals(EXPECTED_200_MSG, 200, statusCode);
+          log.info("\tKeystore reload test : Valid connection found");
+        }
+        catch (Exception e) {
+          fail();
+          // we have to wait for the good key to take affect
+        }
+      });
+
     } finally {
       if (app != null) {
         app.stop();
@@ -271,6 +322,45 @@ public void testHttpsWithNoClientCertAndNoServerTruststore() throws Exception {
     }
   }
 
+  @Test(expected = IOException.class)
+  public void testHttpsWithEmptyStringTruststorePassword() throws Exception {
+    Properties props = new Properties();
+    String uri = "https://localhost:8080";
+    props.put(RestConfig.LISTENERS_CONFIG, uri);
+    configServerKeystore(props);
+    configServerTruststore(props, "");
+    TestRestConfig config = new TestRestConfig(props);
+    SslTestApplication app = new SslTestApplication(config);
+    try {
+      // Empty string is a valid password, but it's not the password the truststore uses
+      // The app should fail at startup with:
+      // java.io.IOException: Keystore was tampered with, or password was incorrect
+      app.start();
+    } finally {
+      app.stop();
+    }
+  }
+
+  @Test
+  public void testHttpsWithNoTruststorePassword() throws Exception {
+    Properties props = new Properties();
+    String uri = "https://localhost:8080";
+    props.put(RestConfig.LISTENERS_CONFIG, uri);
+    configServerKeystore(props);
+    configServerNoTruststorePassword(props);
+    TestRestConfig config = new TestRestConfig(props);
+    SslTestApplication app = new SslTestApplication(config);
+    try {
+      // With no password set (null), verification of the truststore is disabled
+      app.start();
+
+      int statusCode = makeGetRequest(uri + "/test");
+      assertEquals(EXPECTED_200_MSG, 200, statusCode);
+    } finally {
+      app.stop();
+    }
+  }
+
   @Test(expected = SocketException.class)
   public void testHttpsWithAuthAndBadClientCert() throws Exception {
     Properties props = new Properties();
@@ -285,7 +375,7 @@ public void testHttpsWithAuthAndBadClientCert() throws Exception {
       app.start();
 
       // create a new client cert that isn't in the server's trust store.
-      File untrustedClient = File.createTempFile("SslTest-client-keystore", ".jks");
+      File untrustedClient = File.createTempFile("SslTest-client-keystore", ".jks", tempFolder.getRoot());
       Map<String, X509Certificate> certs = new HashMap<>();
       createKeystoreWithCert(untrustedClient, "client", certs);
       try {
diff --git a/pom.xml b/pom.xml
index b41764f03f..60cb18e37d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -55,6 +55,7 @@
         <asynchttpclient.version>2.2.0</asynchttpclient.version>
         <guava.version>24.0-jre</guava.version>
         <checkstyle.suppressions.location>checkstyle/suppressions.xml</checkstyle.suppressions.location>
+        <awaitility.version>4.0.3</awaitility.version>
     </properties>
 
     <repositories>
@@ -182,6 +183,12 @@
                 <artifactId>jersey-test-framework-provider-jetty</artifactId>
                 <version>${jersey.version}</version>
             </dependency>
+
+            <dependency>
+                <groupId>org.awaitility</groupId>
+                <artifactId>awaitility</artifactId>
+                <version>${awaitility.version}</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>