diff --git a/.travis.yml b/.travis.yml
index d35d5e3..1dcdc0e 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -43,9 +43,9 @@
   - sudo add-apt-repository ppa:deadsnakes/ppa -y
   - sudo apt-get update
   - sudo apt-get install python3.6
-  - wget http://mirrors.rackhosting.com/apache/maven/maven-3/3.6.1/binaries/apache-maven-3.6.1-bin.tar.gz -P $HOME
-  - tar xzvf $HOME/apache-maven-3.6.1-bin.tar.gz -C $HOME
-  - export PATH=$HOME/apache-maven-3.6.1/bin:$PATH
+  - export MVN_HOME=$HOME/apache-maven-3.6.1
+  - if [ ! -d $MVN_HOME/bin ]; then wget https://archive.apache.org/dist/maven/maven-3/3.6.1/binaries/apache-maven-3.6.1-bin.tar.gz -P $HOME; tar xzvf $HOME/apache-maven-3.6.1-bin.tar.gz -C $HOME; fi
+  - export PATH=$MVN_HOME/bin:$PATH
 install: /bin/bash ./dev-tools/travis/travis-install.sh `pwd`
 script:
   - /bin/bash ./dev-tools/travis/travis-script.sh `pwd` $MODULES
@@ -54,3 +54,4 @@
     - "$HOME/.m2/repository"
     - "$HOME/.rvm"
     - "$NVM_DIR"
+    - "$HOME/apache-maven-3.6.1"
diff --git a/docs/Setting-up-a-Storm-cluster.md b/docs/Setting-up-a-Storm-cluster.md
index ea05da4..b57639e 100644
--- a/docs/Setting-up-a-Storm-cluster.md
+++ b/docs/Setting-up-a-Storm-cluster.md
@@ -30,7 +30,7 @@
 Next you need to install Storm's dependencies on Nimbus and the worker machines. These are:
 
 1. Java 8+ (Apache Storm 2.x is tested through travis ci against a java 8 JDK)
-2. Python 2.6.6 (Python 3.x should work too, but is not tested as part of our CI enviornment)
+2. Python 2.7.x or Python 3.x
 
 These are the versions of the dependencies that have been tested with Storm. Storm may or may not work with different versions of Java and/or Python.
 
diff --git a/docs/flux.md b/docs/flux.md
index 27d28bb..000270f 100644
--- a/docs/flux.md
+++ b/docs/flux.md
@@ -47,7 +47,7 @@
 If you would like to build Flux from source and run the unit/integration tests, you will need the following installed
 on your system:
 
-* Python 2.6.x or later
+* Python 2.7.x or later
 * Node.js 0.10.x or later
 
 #### Building with unit tests enabled:
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyClient.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyClient.java
index c4d15fc..31425d4 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyClient.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyClient.java
@@ -23,7 +23,6 @@
 import javax.security.auth.callback.CallbackHandler;
 import javax.security.auth.callback.UnsupportedCallbackException;
 import javax.security.auth.kerberos.KerberosTicket;
-import javax.security.auth.login.Configuration;
 import javax.security.auth.login.LoginException;
 import javax.security.sasl.Sasl;
 import javax.security.sasl.SaslClient;
@@ -55,26 +54,16 @@
                   SaslUtils.KERBEROS);
 
         LOG.info("Creating Kerberos Client.");
-
-        Configuration loginConf;
-        try {
-            loginConf = ClientAuthUtils.getConfiguration(topoConf);
-        } catch (Throwable t) {
-            LOG.error("Failed to get loginConf: ", t);
-            throw t;
-        }
         LOG.debug("KerberosSaslNettyClient: authmethod {}", SaslUtils.KERBEROS);
 
         SaslClientCallbackHandler ch = new SaslClientCallbackHandler();
 
+        String jaasConfFile = ClientAuthUtils.getJaasConf(topoConf);
+
         subject = null;
         try {
-            LOG.debug("Setting Configuration to login_config: {}", loginConf);
-            //specify a configuration object to be used
-            Configuration.setConfiguration(loginConf);
-            //now login
-            LOG.debug("Trying to login.");
-            Login login = new Login(jaasSection, ch);
+            LOG.debug("Trying to login using {}.", jaasConfFile);
+            Login login = new Login(jaasSection, ch, jaasConfFile);
             subject = login.getSubject();
             LOG.debug("Got Subject: {}", subject.toString());
         } catch (LoginException ex) {
@@ -88,12 +77,12 @@
             throw new RuntimeException("Fail to verify user principal with section \""
                     + jaasSection
                     + "\" in login configuration file "
-                    + loginConf);
+                    + jaasConfFile);
         }
 
         String serviceName = null;
         try {
-            serviceName = ClientAuthUtils.get(loginConf, jaasSection, "serviceName");
+            serviceName = ClientAuthUtils.get(topoConf, jaasSection, "serviceName");
         } catch (IOException e) {
             LOG.error("Failed to get service name.", e);
             throw new RuntimeException(e);
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyServer.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyServer.java
index ee410d6..fefcdc6 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyServer.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/KerberosSaslNettyServer.java
@@ -25,7 +25,6 @@
 import javax.security.auth.callback.UnsupportedCallbackException;
 import javax.security.auth.kerberos.KerberosPrincipal;
 import javax.security.auth.kerberos.KerberosTicket;
-import javax.security.auth.login.Configuration;
 import javax.security.auth.login.LoginException;
 import javax.security.sasl.AuthorizeCallback;
 import javax.security.sasl.Sasl;
@@ -49,28 +48,17 @@
 
     KerberosSaslNettyServer(Map<String, Object> topoConf, String jaasSection, List<String> authorizedUsers) {
         this.authorizedUsers = authorizedUsers;
-        LOG.debug("Getting Configuration.");
-        Configuration loginConf;
-        try {
-            loginConf = ClientAuthUtils.getConfiguration(topoConf);
-        } catch (Throwable t) {
-            LOG.error("Failed to get loginConf: ", t);
-            throw t;
-        }
 
         LOG.debug("KerberosSaslNettyServer: authmethod {}", SaslUtils.KERBEROS);
 
         KerberosSaslCallbackHandler ch = new KerberosSaslNettyServer.KerberosSaslCallbackHandler(authorizedUsers);
+        String jaasConfFile = ClientAuthUtils.getJaasConf(topoConf);
 
         //login our principal
         subject = null;
         try {
-            LOG.debug("Setting Configuration to login_config: {}", loginConf);
-            //specify a configuration object to be used
-            Configuration.setConfiguration(loginConf);
-            //now login
-            LOG.debug("Trying to login.");
-            Login login = new Login(jaasSection, ch);
+            LOG.debug("Trying to login using {}.", jaasConfFile);
+            Login login = new Login(jaasSection, ch, jaasConfFile);
             subject = login.getSubject();
             LOG.debug("Got Subject: {}", subject.toString());
         } catch (LoginException ex) {
@@ -84,7 +72,7 @@
             throw new RuntimeException("Fail to verify user principal with section \""
                                        + jaasSection
                                        + "\" in login configuration file "
-                                       + loginConf);
+                                       + jaasConfFile);
         }
 
         try {
diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/Login.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/Login.java
index 6a50d84..d6a345e 100644
--- a/storm-client/src/jvm/org/apache/storm/messaging/netty/Login.java
+++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/Login.java
@@ -19,6 +19,9 @@
  * does not die.
  */
 
+import java.io.File;
+import java.net.URI;
+import java.security.URIParameter;
 import java.util.Date;
 import java.util.Random;
 import java.util.Set;
@@ -31,6 +34,7 @@
 import javax.security.auth.login.LoginContext;
 import javax.security.auth.login.LoginException;
 import org.apache.log4j.Logger;
+import org.apache.storm.security.auth.ClientAuthUtils;
 import org.apache.storm.shade.org.apache.zookeeper.Shell;
 import org.apache.storm.shade.org.apache.zookeeper.client.ZooKeeperSaslClient;
 
@@ -57,12 +61,9 @@
     private Thread thread = null;
     private boolean isKrbTicket = false;
     private boolean isUsingTicketCache = false;
-    private boolean isUsingKeytab = false;
     private LoginContext login = null;
     private String loginContextName = null;
-    private String keytabFile = null;
     private String principal = null;
-
     private long lastLogin = 0;
 
     /**
@@ -77,14 +78,14 @@
      * @throws javax.security.auth.login.LoginException
      *               Thrown if authentication fails.
      */
-    public Login(final String loginContextName, CallbackHandler callbackHandler)
+    public Login(final String loginContextName, CallbackHandler callbackHandler, String jaasConfFile)
         throws LoginException {
         this.callbackHandler = callbackHandler;
-        login = login(loginContextName);
+        login = login(loginContextName, jaasConfFile);
         this.loginContextName = loginContextName;
         subject = login.getSubject();
         isKrbTicket = !subject.getPrivateCredentials(KerberosTicket.class).isEmpty();
-        AppConfigurationEntry[] entries = Configuration.getConfiguration().getAppConfigurationEntry(loginContextName);
+        AppConfigurationEntry[] entries = this.getConfiguration(jaasConfFile).getAppConfigurationEntry(loginContextName);
         for (AppConfigurationEntry entry : entries) {
             // there will only be a single entry, so this for() loop will only be iterated through once.
             if (entry.getOptions().get("useTicketCache") != null) {
@@ -93,10 +94,6 @@
                     isUsingTicketCache = true;
                 }
             }
-            if (entry.getOptions().get("keyTab") != null) {
-                keytabFile = (String) entry.getOptions().get("keyTab");
-                isUsingKeytab = true;
-            }
             if (entry.getOptions().get("principal") != null) {
                 principal = (String) entry.getOptions().get("principal");
             }
@@ -251,6 +248,19 @@
         thread.setDaemon(true);
     }
 
+    private Configuration getConfiguration(String jaasConfFile) {
+        File configFile = new File(jaasConfFile);
+        if (!configFile.canRead()) {
+            throw new RuntimeException("File " + jaasConfFile + " cannot be read.");
+        }
+        try {
+            URI configUri = configFile.toURI();
+            return Configuration.getInstance("JavaLoginConfig", new URIParameter(configUri));
+        } catch (Exception ex) {
+            throw new RuntimeException("Failed to get configuration for " + jaasConfFile, ex);
+        }
+    }
+
     public void startThreadIfNeeded() {
         // thread object 'thread' will be null if a refresh thread is not needed.
         if (thread != null) {
@@ -277,7 +287,7 @@
         return loginContextName;
     }
 
-    private synchronized LoginContext login(final String loginContextName) throws LoginException {
+    private synchronized LoginContext login(final String loginContextName, String jaasConfFile) throws LoginException {
         if (loginContextName == null) {
             throw new LoginException("loginContext name (JAAS file section header) was null. "
                     + "Please check your java.security.login.auth.config (="
@@ -285,9 +295,10 @@
                     + ") and your " + ZooKeeperSaslClient.LOGIN_CONTEXT_NAME_KEY + "(="
                     + System.getProperty(ZooKeeperSaslClient.LOGIN_CONTEXT_NAME_KEY, "Client") + ")");
         }
-        LoginContext loginContext = new LoginContext(loginContextName, callbackHandler);
+        Configuration configuration = this.getConfiguration(jaasConfFile);
+        LoginContext loginContext = new LoginContext(loginContextName, null, callbackHandler, configuration);
         loginContext.login();
-        LOG.info("successfully logged in.");
+        LOG.info("Successfully logged in to context " + loginContextName + " using " + jaasConfFile);
         return loginContext;
     }
 
diff --git a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
index b77e003..30faf1f 100644
--- a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
+++ b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
@@ -73,9 +73,8 @@
         switch (auth) {
 
             case "DIGEST":
-                Configuration loginConf = ClientAuthUtils.getConfiguration(config);
                 authMethod = ThriftNettyClientCodec.AuthMethod.DIGEST;
-                secret = ClientAuthUtils.makeDigestPayload(loginConf, ClientAuthUtils.LOGIN_CONTEXT_PACEMAKER_DIGEST);
+                secret = ClientAuthUtils.makeDigestPayload(config, ClientAuthUtils.LOGIN_CONTEXT_PACEMAKER_DIGEST);
                 if (secret == null) {
                     LOG.error("Can't start pacemaker server without digest secret.");
                     throw new RuntimeException("Can't start pacemaker server without digest secret.");
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/ClientAuthUtils.java b/storm-client/src/jvm/org/apache/storm/security/auth/ClientAuthUtils.java
index 9fb0e4b..6a9b703 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/ClientAuthUtils.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/ClientAuthUtils.java
@@ -60,6 +60,10 @@
     private static final String USERNAME = "username";
     private static final String PASSWORD = "password";
 
+    public static String getJaasConf(Map<String, Object> topoConf) {
+        return (String) topoConf.get("java.security.auth.login.config");
+    }
+
     /**
      * Construct a JAAS configuration object per storm configuration file.
      *
@@ -70,7 +74,7 @@
         Configuration loginConf = null;
 
         //find login file configuration from Storm configuration
-        String loginConfigurationFile = (String) topoConf.get("java.security.auth.login.config");
+        String loginConfigurationFile = getJaasConf(topoConf);
         if ((loginConfigurationFile != null) && (loginConfigurationFile.length() > 0)) {
             File configFile = new File(loginConfigurationFile);
             if (!configFile.canRead()) {
@@ -111,12 +115,13 @@
     /**
      * Pull a set of keys out of a Configuration.
      *
-     * @param configuration The config to pull the key/value pairs out of.
+     * @param topoConf  The config containing the jaas conf file.
      * @param section       The app configuration entry name to get stuff from.
      * @return Return a map of the configs in conf.
      */
-    public static SortedMap<String, ?> pullConfig(Configuration configuration,
+    public static SortedMap<String, ?> pullConfig(Map<String, Object> topoConf,
                                                   String section) throws IOException {
+        Configuration configuration = ClientAuthUtils.getConfiguration(topoConf);
         AppConfigurationEntry[] configurationEntries = ClientAuthUtils.getEntries(configuration, section);
 
         if (configurationEntries == null) {
@@ -138,12 +143,17 @@
     /**
      * Pull a the value given section and key from Configuration.
      *
-     * @param configuration The config to pull the key/value pairs out of.
+     * @param topoConf   The config containing the jaas conf file.
      * @param section       The app configuration entry name to get stuff from.
      * @param key           The key to look up inside of the section
      * @return Return a the String value of the configuration value
      */
-    public static String get(Configuration configuration, String section, String key) throws IOException {
+    public static String get(Map<String, Object> topoConf, String section, String key) throws IOException {
+        Configuration configuration = ClientAuthUtils.getConfiguration(topoConf);
+        return get(configuration, section, key);
+    }
+
+    static String get(Configuration configuration, String section, String key) throws IOException {
         AppConfigurationEntry[] configurationEntries = ClientAuthUtils.getEntries(configuration, section);
 
         if (configurationEntries == null) {
@@ -456,22 +466,22 @@
     /**
      * Construct a transport plugin per storm configuration.
      */
-    public static ITransportPlugin getTransportPlugin(ThriftConnectionType type, Map<String, Object> topoConf, Configuration loginConf) {
+    public static ITransportPlugin getTransportPlugin(ThriftConnectionType type, Map<String, Object> topoConf) {
         try {
             String transportPluginClassName = type.getTransportPlugin(topoConf);
             ITransportPlugin transportPlugin = ReflectionUtils.newInstance(transportPluginClassName);
-            transportPlugin.prepare(type, topoConf, loginConf);
+            transportPlugin.prepare(type, topoConf);
             return transportPlugin;
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
     }
 
-    public static String makeDigestPayload(Configuration loginConfig, String configSection) {
+    public static String makeDigestPayload(Map<String, Object> topoConf, String configSection) {
         String username = null;
         String password = null;
         try {
-            Map<String, ?> results = ClientAuthUtils.pullConfig(loginConfig, configSection);
+            Map<String, ?> results = ClientAuthUtils.pullConfig(topoConf, configSection);
             username = (String) results.get(USERNAME);
             password = (String) results.get(PASSWORD);
         } catch (Exception e) {
@@ -492,6 +502,8 @@
         }
     }
 
+
+
     public static byte[] serializeKerberosTicket(KerberosTicket tgt) throws Exception {
         ByteArrayOutputStream bao = new ByteArrayOutputStream();
         ObjectOutputStream out = new ObjectOutputStream(bao);
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/ITransportPlugin.java b/storm-client/src/jvm/org/apache/storm/security/auth/ITransportPlugin.java
index 6bf3b0b..19ec374 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/ITransportPlugin.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/ITransportPlugin.java
@@ -29,9 +29,8 @@
      *
      * @param type      the type of connection this will process.
      * @param topoConf  Storm configuration
-     * @param loginConf login configuration
      */
-    void prepare(ThriftConnectionType type, Map<String, Object> topoConf, Configuration loginConf);
+    void prepare(ThriftConnectionType type, Map<String, Object> topoConf);
 
     /**
      * Create a server associated with a given port, service handler, and purpose.
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java b/storm-client/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java
index 0276525..577f9fd 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/SimpleTransportPlugin.java
@@ -48,14 +48,12 @@
     private static final Logger LOG = LoggerFactory.getLogger(SimpleTransportPlugin.class);
     protected ThriftConnectionType type;
     protected Map<String, Object> topoConf;
-    protected Configuration loginConf;
     private int port;
 
     @Override
-    public void prepare(ThriftConnectionType type, Map<String, Object> topoConf, Configuration loginConf) {
+    public void prepare(ThriftConnectionType type, Map<String, Object> topoConf) {
         this.type = type;
         this.topoConf = topoConf;
-        this.loginConf = loginConf;
     }
 
     @Override
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/ThriftClient.java b/storm-client/src/jvm/org/apache/storm/security/auth/ThriftClient.java
index a9d9129..88d70ec 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/ThriftClient.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/ThriftClient.java
@@ -83,11 +83,8 @@
                 socket.setTimeout(timeout);
             }
 
-            //locate login configuration 
-            Configuration loginConf = ClientAuthUtils.getConfiguration(conf);
-
             //construct a transport plugin
-            ITransportPlugin transportPlugin = ClientAuthUtils.getTransportPlugin(type, conf, loginConf);
+            ITransportPlugin transportPlugin = ClientAuthUtils.getTransportPlugin(type, conf);
 
             //TODO get this from type instead of hardcoding to Nimbus.
             //establish client-server transport via plugin
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/ThriftServer.java b/storm-client/src/jvm/org/apache/storm/security/auth/ThriftServer.java
index f6102bc..9938b7c 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/ThriftServer.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/ThriftServer.java
@@ -12,10 +12,8 @@
 
 package org.apache.storm.security.auth;
 
-import java.io.Closeable;
 import java.io.IOException;
 import java.util.Map;
-import javax.security.auth.login.Configuration;
 import org.apache.storm.security.auth.sasl.SaslTransportPlugin;
 import org.apache.storm.thrift.TProcessor;
 import org.apache.storm.thrift.server.TServer;
@@ -29,7 +27,6 @@
     private final Map<String, Object> conf; //storm configuration
     private final ThriftConnectionType type;
     private TServer server;
-    private Configuration loginConf;
     private int port;
     private boolean areWorkerTokensSupported;
     private ITransportPlugin transportPlugin;
@@ -40,14 +37,8 @@
         this.type = type;
 
         try {
-            //retrieve authentication configuration 
-            loginConf = ClientAuthUtils.getConfiguration(this.conf);
-        } catch (Exception x) {
-            LOG.error(x.getMessage(), x);
-        }
-        try {
             //locate our thrift transport plugin
-            transportPlugin = ClientAuthUtils.getTransportPlugin(this.type, this.conf, loginConf);
+            transportPlugin = ClientAuthUtils.getTransportPlugin(this.type, this.conf);
             //server
             server = transportPlugin.getServer(this.processor);
             port = transportPlugin.getPort();
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/digest/DigestSaslTransportPlugin.java b/storm-client/src/jvm/org/apache/storm/security/auth/digest/DigestSaslTransportPlugin.java
index e3e4497..5a77d51 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/digest/DigestSaslTransportPlugin.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/digest/DigestSaslTransportPlugin.java
@@ -16,6 +16,8 @@
 import java.util.Map;
 import javax.security.auth.callback.CallbackHandler;
 import javax.security.auth.login.AppConfigurationEntry;
+import javax.security.auth.login.Configuration;
+
 import org.apache.storm.generated.WorkerToken;
 import org.apache.storm.security.auth.ClientAuthUtils;
 import org.apache.storm.security.auth.sasl.SaslTransportPlugin;
@@ -44,7 +46,7 @@
         //create an authentication callback handler
         CallbackHandler serverCallbackHandler = new SimpleSaslServerCallbackHandler(impersonationAllowed,
                                                                                     workerTokenAuthorizer,
-                                                                                    new JassPasswordProvider(loginConf));
+                                                                                    new JassPasswordProvider(conf));
 
         //create a transport factory that will invoke our auth callback for digest
         TSaslServerTransport.Factory factory = new TSaslServerTransport.Factory();
@@ -60,7 +62,11 @@
         WorkerToken token = WorkerTokenClientCallbackHandler.findWorkerTokenInSubject(type);
         if (token != null) {
             clientCallbackHandler = new WorkerTokenClientCallbackHandler(token);
-        } else if (loginConf != null) {
+        } else {
+            Configuration loginConf = ClientAuthUtils.getConfiguration(conf);
+            if (loginConf == null) {
+                throw new IOException("Could not find any way to authenticate with the server.");
+            }
             AppConfigurationEntry[] configurationEntries = loginConf.getAppConfigurationEntry(ClientAuthUtils.LOGIN_CONTEXT_CLIENT);
             if (configurationEntries == null) {
                 String errorMessage = "Could not find a '" + ClientAuthUtils.LOGIN_CONTEXT_CLIENT
@@ -76,8 +82,6 @@
                 password = (String) options.getOrDefault("password", password);
             }
             clientCallbackHandler = new SimpleSaslClientCallbackHandler(username, password);
-        } else {
-            throw new IOException("Could not find any way to authenticate with the server.");
         }
 
         TSaslClientTransport wrapperTransport = new TSaslClientTransport(DIGEST,
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/digest/JassPasswordProvider.java b/storm-client/src/jvm/org/apache/storm/security/auth/digest/JassPasswordProvider.java
index 5c0d7a4..1d5c7e3 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/digest/JassPasswordProvider.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/digest/JassPasswordProvider.java
@@ -36,10 +36,12 @@
     /**
      * Constructor.
      *
-     * @param configuration the jaas configuration to get the credentials out of.
+     * @param topoConf the configuration containing the jaas conf to use.
      * @throws IOException if we could not read the Server section in the jaas conf.
      */
-    public JassPasswordProvider(Configuration configuration) throws IOException {
+    public JassPasswordProvider(Map<String, Object> topoConf) throws IOException {
+
+        Configuration configuration = ClientAuthUtils.getConfiguration(topoConf);
         if (configuration == null) {
             return;
         }
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/AutoTGT.java b/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/AutoTGT.java
index c4f9f91..6936555 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/AutoTGT.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/AutoTGT.java
@@ -121,11 +121,10 @@
         //Log the user in and get the TGT
         try {
             Configuration loginConf = ClientAuthUtils.getConfiguration(conf);
-            ClientCallbackHandler clientCallbackHandler = new ClientCallbackHandler(loginConf);
+            ClientCallbackHandler clientCallbackHandler = new ClientCallbackHandler(conf);
 
             //login our user
-            Configuration.setConfiguration(loginConf);
-            LoginContext lc = new LoginContext(ClientAuthUtils.LOGIN_CONTEXT_CLIENT, clientCallbackHandler);
+            LoginContext lc = new LoginContext(ClientAuthUtils.LOGIN_CONTEXT_CLIENT, null, clientCallbackHandler, loginConf);
             try {
                 lc.login();
                 final Subject subject = lc.getSubject();
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/ClientCallbackHandler.java b/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/ClientCallbackHandler.java
index 64673ce..f9a8e09 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/ClientCallbackHandler.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/ClientCallbackHandler.java
@@ -13,6 +13,7 @@
 package org.apache.storm.security.auth.kerberos;
 
 import java.io.IOException;
+import java.util.Map;
 import javax.security.auth.callback.Callback;
 import javax.security.auth.callback.CallbackHandler;
 import javax.security.auth.callback.NameCallback;
@@ -36,7 +37,8 @@
      *
      * <p>For digest, you should have a pair of user name and password defined in this figgure.
      */
-    public ClientCallbackHandler(Configuration configuration) throws IOException {
+    public ClientCallbackHandler(Map<String, Object> topoConf) throws IOException {
+        Configuration configuration = ClientAuthUtils.getConfiguration(topoConf);
         if (configuration == null) {
             return;
         }
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/KerberosSaslTransportPlugin.java b/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/KerberosSaslTransportPlugin.java
index 915473b..5b42886 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/KerberosSaslTransportPlugin.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/KerberosSaslTransportPlugin.java
@@ -58,15 +58,15 @@
             workerTokenAuthorizer = new WorkerTokenAuthorizer(conf, type);
         }
         //create an authentication callback handler
-        CallbackHandler serverCallbackHandler = new ServerCallbackHandler(loginConf, impersonationAllowed);
+        CallbackHandler serverCallbackHandler = new ServerCallbackHandler(conf, impersonationAllowed);
+
+        String jaasConfFile = ClientAuthUtils.getJaasConf(conf);
 
         //login our principal
         Subject subject = null;
         try {
-            //specify a configuration object to be used
-            Configuration.setConfiguration(loginConf);
             //now login
-            Login login = new Login(ClientAuthUtils.LOGIN_CONTEXT_SERVER, serverCallbackHandler);
+            Login login = new Login(ClientAuthUtils.LOGIN_CONTEXT_SERVER, serverCallbackHandler, jaasConfFile);
             subject = login.getSubject();
             login.startThreadIfNeeded();
         } catch (LoginException ex) {
@@ -77,10 +77,10 @@
         //check the credential of our principal
         if (subject.getPrivateCredentials(KerberosTicket.class).isEmpty()) {
             throw new RuntimeException("Fail to verify user principal with section \""
-                                       + ClientAuthUtils.LOGIN_CONTEXT_SERVER + "\" in login configuration file " + loginConf);
+                                       + ClientAuthUtils.LOGIN_CONTEXT_SERVER + "\" in login configuration file " + jaasConfFile);
         }
 
-        String principal = ClientAuthUtils.get(loginConf, ClientAuthUtils.LOGIN_CONTEXT_SERVER, "principal");
+        String principal = ClientAuthUtils.get(conf, ClientAuthUtils.LOGIN_CONTEXT_SERVER, "principal");
         LOG.debug("principal:" + principal);
         KerberosName serviceKerberosName = new KerberosName(principal);
         String serviceName = serviceKerberosName.getServiceName();
@@ -107,11 +107,9 @@
     private Login mkLogin() throws IOException {
         try {
             //create an authentication callback handler
-            ClientCallbackHandler clientCallbackHandler = new ClientCallbackHandler(loginConf);
-            //specify a configuration object to be used
-            Configuration.setConfiguration(loginConf);
+            ClientCallbackHandler clientCallbackHandler = new ClientCallbackHandler(conf);
             //now login
-            Login login = new Login(ClientAuthUtils.LOGIN_CONTEXT_CLIENT, clientCallbackHandler);
+            Login login = new Login(ClientAuthUtils.LOGIN_CONTEXT_CLIENT, clientCallbackHandler, ClientAuthUtils.getJaasConf(conf));
             login.startThreadIfNeeded();
             return login;
         } catch (LoginException ex) {
@@ -142,7 +140,7 @@
 
     private TTransport kerberosConnect(TTransport transport, String serverHost, String asUser) throws IOException {
         //login our user
-        SortedMap<String, ?> authConf = ClientAuthUtils.pullConfig(loginConf, ClientAuthUtils.LOGIN_CONTEXT_CLIENT);
+        SortedMap<String, ?> authConf = ClientAuthUtils.pullConfig(conf, ClientAuthUtils.LOGIN_CONTEXT_CLIENT);
         if (authConf == null) {
             throw new RuntimeException("Error in parsing the kerberos login Configuration, returned null");
         }
@@ -180,11 +178,11 @@
         final Subject subject = login.getSubject();
         if (subject.getPrivateCredentials(KerberosTicket.class).isEmpty()) { //error
             throw new RuntimeException("Fail to verify user principal with section \""
-                                       + ClientAuthUtils.LOGIN_CONTEXT_CLIENT + "\" in login configuration file " + loginConf);
+                    + ClientAuthUtils.LOGIN_CONTEXT_CLIENT + "\" in login configuration file " + ClientAuthUtils.getJaasConf(conf));
         }
 
         final String principal = StringUtils.isBlank(asUser) ? getPrincipal(subject) : asUser;
-        String serviceName = ClientAuthUtils.get(loginConf, ClientAuthUtils.LOGIN_CONTEXT_CLIENT, "serviceName");
+        String serviceName = ClientAuthUtils.get(conf, ClientAuthUtils.LOGIN_CONTEXT_CLIENT, "serviceName");
         if (serviceName == null) {
             serviceName = ClientAuthUtils.SERVICE;
         }
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/ServerCallbackHandler.java b/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/ServerCallbackHandler.java
index 05630e9..daa5fbf 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/ServerCallbackHandler.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/kerberos/ServerCallbackHandler.java
@@ -13,6 +13,7 @@
 package org.apache.storm.security.auth.kerberos;
 
 import java.io.IOException;
+import java.util.Map;
 import javax.security.auth.callback.Callback;
 import javax.security.auth.callback.CallbackHandler;
 import javax.security.auth.callback.NameCallback;
@@ -35,8 +36,10 @@
     private static final Logger LOG = LoggerFactory.getLogger(ServerCallbackHandler.class);
     private final boolean impersonationAllowed;
 
-    public ServerCallbackHandler(Configuration configuration, boolean impersonationAllowed) throws IOException {
+    public ServerCallbackHandler(Map<String, Object> topoConf, boolean impersonationAllowed) throws IOException {
         this.impersonationAllowed = impersonationAllowed;
+
+        Configuration configuration = ClientAuthUtils.getConfiguration(topoConf);
         if (configuration == null) {
             return;
         }
diff --git a/storm-client/src/jvm/org/apache/storm/security/auth/sasl/SaslTransportPlugin.java b/storm-client/src/jvm/org/apache/storm/security/auth/sasl/SaslTransportPlugin.java
index 343c247..745d360 100644
--- a/storm-client/src/jvm/org/apache/storm/security/auth/sasl/SaslTransportPlugin.java
+++ b/storm-client/src/jvm/org/apache/storm/security/auth/sasl/SaslTransportPlugin.java
@@ -49,14 +49,12 @@
 public abstract class SaslTransportPlugin implements ITransportPlugin, Closeable {
     protected ThriftConnectionType type;
     protected Map<String, Object> conf;
-    protected Configuration loginConf;
     private int port;
 
     @Override
-    public void prepare(ThriftConnectionType type, Map<String, Object> conf, Configuration loginConf) {
+    public void prepare(ThriftConnectionType type, Map<String, Object> conf) {
         this.type = type;
         this.conf = conf;
-        this.loginConf = loginConf;
     }
 
     @Override
diff --git a/storm-client/test/jvm/org/apache/storm/daemon/worker/LogConfigManagerTest.java b/storm-client/test/jvm/org/apache/storm/daemon/worker/LogConfigManagerTest.java
index bf8ded8..673458f 100644
--- a/storm-client/test/jvm/org/apache/storm/daemon/worker/LogConfigManagerTest.java
+++ b/storm-client/test/jvm/org/apache/storm/daemon/worker/LogConfigManagerTest.java
@@ -14,6 +14,7 @@
 
 import java.util.TreeMap;
 import java.util.concurrent.atomic.AtomicReference;
+
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.core.LoggerContext;
 import org.apache.storm.generated.LogConfig;
@@ -27,6 +28,7 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.Mockito.anyObject;
 import static org.mockito.Mockito.anyString;
 import static org.mockito.Mockito.eq;
@@ -203,6 +205,32 @@
     }
 
     @Test
+    public void testProcessLogConfigChangeThrowsNullPointerExceptionWhenTargetLogLevelIsNotSet() {
+        LogConfigManager logConfigManager = new LogConfigManager();
+
+        LogConfig logConfig = new LogConfig();
+        LogLevel logLevel = new LogLevel();
+        logLevel.set_action(LogLevelAction.UPDATE);
+        logLevel.set_reset_log_level("INFO");
+        logConfig.put_to_named_logger_level("RESET_LOG", logLevel);
+
+        assertThrows(NullPointerException.class, () -> logConfigManager.processLogConfigChange(logConfig));
+    }
+
+    @Test
+    public void testProcessLogConfigChangeExecutesSuccessfullyWhenTargetLogLevelIsSet() {
+        LogConfigManager logConfigManager = new LogConfigManager();
+
+        LogConfig logConfig = new LogConfig();
+        LogLevel logLevel = new LogLevel();
+        logLevel.set_action(LogLevelAction.UPDATE);
+        logLevel.set_target_log_level("DEBUG");
+        logConfig.put_to_named_logger_level("TARGET_LOG", logLevel);
+
+        logConfigManager.processLogConfigChange(logConfig);
+    }
+
+    @Test
     public void testProcessRootLogLevelToDebugSetsLoggerAndTimeout() {
         try (SimulatedTime t = new SimulatedTime()) {
             LogConfig mockConfig = new LogConfig();
diff --git a/storm-client/test/jvm/org/apache/storm/security/auth/ClientAuthUtilsTest.java b/storm-client/test/jvm/org/apache/storm/security/auth/ClientAuthUtilsTest.java
index 9f09ee4..1aa854b 100644
--- a/storm-client/test/jvm/org/apache/storm/security/auth/ClientAuthUtilsTest.java
+++ b/storm-client/test/jvm/org/apache/storm/security/auth/ClientAuthUtilsTest.java
@@ -95,8 +95,9 @@
 
     @Test
     public void objGettersReturnNullWithNullConfigTest() throws IOException {
-        Assert.assertNull(ClientAuthUtils.pullConfig(null, "foo"));
-        Assert.assertNull(ClientAuthUtils.get(null, "foo", "bar"));
+        Map<String, Object> topoConf = new HashMap<>();
+        Assert.assertNull(ClientAuthUtils.pullConfig(topoConf, "foo"));
+        Assert.assertNull(ClientAuthUtils.get(topoConf, "foo", "bar"));
 
         Assert.assertNull(ClientAuthUtils.getConfiguration(Collections.emptyMap()));
     }
@@ -141,39 +142,6 @@
         Mockito.verify(autoCred, Mockito.times(1)).populateSubject(subject, cred);
     }
 
-    @Test
-    public void makeDigestPayloadTest() throws NoSuchAlgorithmException {
-        String section = "user-pass-section";
-        Map<String, String> optionMap = new HashMap<String, String>();
-        String user = "user";
-        String pass = "pass";
-        optionMap.put("username", user);
-        optionMap.put("password", pass);
-        AppConfigurationEntry entry = Mockito.mock(AppConfigurationEntry.class);
-
-        Mockito.<Map<String, ?>>when(entry.getOptions()).thenReturn(optionMap);
-        Configuration mockConfig = Mockito.mock(Configuration.class);
-        Mockito.when(mockConfig.getAppConfigurationEntry(section))
-               .thenReturn(new AppConfigurationEntry[]{ entry });
-
-        MessageDigest digest = MessageDigest.getInstance("SHA-512");
-        byte[] output = digest.digest((user + ":" + pass).getBytes());
-        String sha = Hex.encodeHexString(output);
-
-        // previous code used this method to generate the string, ensure the two match
-        StringBuilder builder = new StringBuilder();
-        for (byte b : output) {
-            builder.append(String.format("%02x", b));
-        }
-        String stringFormatMethod = builder.toString();
-
-        Assert.assertEquals(
-            ClientAuthUtils.makeDigestPayload(mockConfig, "user-pass-section"),
-            sha);
-
-        Assert.assertEquals(sha, stringFormatMethod);
-    }
-
     @Test(expected = RuntimeException.class)
     public void invalidConfigResultsInIOException() throws RuntimeException {
         HashMap<String, Object> conf = new HashMap<>();
diff --git a/storm-core/src/jvm/org/apache/storm/command/SetLogLevel.java b/storm-core/src/jvm/org/apache/storm/command/SetLogLevel.java
index 411c586..c6e2fa7 100644
--- a/storm-core/src/jvm/org/apache/storm/command/SetLogLevel.java
+++ b/storm-core/src/jvm/org/apache/storm/command/SetLogLevel.java
@@ -92,7 +92,7 @@
                 splits = splits[1].split(":");
                 Integer timeout = 0;
                 Level level = Level.valueOf(splits[0]);
-                logLevel.set_reset_log_level(level.toString());
+                logLevel.set_target_log_level(level.toString());
                 if (splits.length > 1) {
                     timeout = Integer.parseInt(splits[1]);
                 }
diff --git a/storm-core/test/jvm/org/apache/storm/command/SetLogLevelTest.java b/storm-core/test/jvm/org/apache/storm/command/SetLogLevelTest.java
index 5951752..af6aefc 100644
--- a/storm-core/test/jvm/org/apache/storm/command/SetLogLevelTest.java
+++ b/storm-core/test/jvm/org/apache/storm/command/SetLogLevelTest.java
@@ -25,11 +25,11 @@
         SetLogLevel.LogLevelsParser logLevelsParser = new SetLogLevel.LogLevelsParser(LogLevelAction.UPDATE);
         LogLevel logLevel = ((Map<String, LogLevel>) logLevelsParser.parse("com.foo.one=warn")).get("com.foo.one");
         Assert.assertEquals(0, logLevel.get_reset_log_level_timeout_secs());
-        Assert.assertEquals("WARN", logLevel.get_reset_log_level());
+        Assert.assertEquals("WARN", logLevel.get_target_log_level());
 
         logLevel = ((Map<String, LogLevel>) logLevelsParser.parse("com.foo.two=DEBUG:10")).get("com.foo.two");
         Assert.assertEquals(10, logLevel.get_reset_log_level_timeout_secs());
-        Assert.assertEquals("DEBUG", logLevel.get_reset_log_level());
+        Assert.assertEquals("DEBUG", logLevel.get_target_log_level());
     }
 
     @Test(expected = NumberFormatException.class)
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/supervisor/Slot.java b/storm-server/src/main/java/org/apache/storm/daemon/supervisor/Slot.java
index df419b9..85e5f9a 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/supervisor/Slot.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/supervisor/Slot.java
@@ -12,8 +12,10 @@
 
 package org.apache.storm.daemon.supervisor;
 
+import com.google.common.annotations.VisibleForTesting;
 import java.io.IOException;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -36,13 +38,13 @@
 import org.apache.storm.generated.LocalAssignment;
 import org.apache.storm.generated.ProfileAction;
 import org.apache.storm.generated.ProfileRequest;
+import org.apache.storm.generated.WorkerResources;
 import org.apache.storm.localizer.AsyncLocalizer;
 import org.apache.storm.localizer.BlobChangingCallback;
 import org.apache.storm.localizer.GoodToGo;
 import org.apache.storm.localizer.LocallyCachedBlob;
 import org.apache.storm.metricstore.WorkerMetricsProcessor;
 import org.apache.storm.scheduler.ISupervisor;
-import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
 import org.apache.storm.utils.EnumUtil;
 import org.apache.storm.utils.LocalState;
 import org.apache.storm.utils.ObjectReader;
@@ -174,9 +176,80 @@
     }
 
     /**
-     * Decide the equivalence of two local assignments, ignoring the order of executors
-     * This is different from #equal method.
-     * @param first Local assignment A
+     * This method compares WorkerResources while considering any resources are NULL to be 0.0
+     *
+     * @param first  WorkerResources A
+     * @param second WorkerResources B
+     * @return True if A and B are equivalent, treating the absent resources as 0.0
+     */
+    @VisibleForTesting
+    static boolean customWorkerResourcesEquality(WorkerResources first, WorkerResources second) {
+        if (first == null) {
+            return false;
+        }
+        if (first == second) {
+            return true;
+        }
+        if (first.equals(second)) {
+            return true;
+        }
+
+        if (first.get_cpu() != second.get_cpu()) {
+            return false;
+        }
+        if (first.get_mem_on_heap() != second.get_mem_on_heap()) {
+            return false;
+        }
+        if (first.get_mem_off_heap() != second.get_mem_off_heap()) {
+            return false;
+        }
+        if (first.get_shared_mem_off_heap() != second.get_shared_mem_off_heap()) {
+            return false;
+        }
+        if (first.get_shared_mem_on_heap() != second.get_shared_mem_on_heap()) {
+            return false;
+        }
+        if (!customResourceMapEquality(first.get_resources(), second.get_resources())) {
+            return false;
+        }
+        if (!customResourceMapEquality(first.get_shared_resources(), second.get_shared_resources())) {
+            return false;
+        }
+        return true;
+    }
+
+    /**
+     * This method compares Resource Maps while considering any resources are NULL to be 0.0
+     *
+     * @param firstMap  Resource Map A
+     * @param secondMap Resource Map B
+     * @return True if A and B are equivalent, treating the absent resources as 0.0
+     */
+    private static boolean customResourceMapEquality(Map<String, Double> firstMap, Map<String, Double> secondMap) {
+        if (firstMap == null && secondMap == null) {
+            return true;
+        }
+        if (firstMap == null) {
+            firstMap = new HashMap<>();
+        }
+        if (secondMap == null) {
+            secondMap = new HashMap<>();
+        }
+
+        Set<String> keys = new HashSet<>(firstMap.keySet());
+        keys.addAll(secondMap.keySet());
+        for (String key : keys) {
+            if (firstMap.getOrDefault(key, 0.0).doubleValue() != secondMap.getOrDefault(key, 0.0).doubleValue()) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    /**
+     * Decide the equivalence of two local assignments, ignoring the order of executors This is different from #equal method.
+     *
+     * @param first  Local assignment A
      * @param second Local assignment B
      * @return True if A and B are equivalent, ignoring the order of the executors
      */
@@ -196,9 +269,9 @@
                         return true;
                     }
                     if (firstHasResources && secondHasResources) {
-                        if (first.get_resources().equals(second.get_resources())) {
-                            return true;
-                        }
+                        WorkerResources firstResources = first.get_resources();
+                        WorkerResources secondResources = second.get_resources();
+                        return customWorkerResourcesEquality(firstResources, secondResources);
                     }
                 }
             }
diff --git a/storm-server/src/main/java/org/apache/storm/localizer/AsyncLocalizer.java b/storm-server/src/main/java/org/apache/storm/localizer/AsyncLocalizer.java
index fdd130b..48ad7e8 100644
--- a/storm-server/src/main/java/org/apache/storm/localizer/AsyncLocalizer.java
+++ b/storm-server/src/main/java/org/apache/storm/localizer/AsyncLocalizer.java
@@ -262,22 +262,26 @@
                     while (!done) {
                         try {
                             synchronized (blob) {
-                                long localVersion = blob.getLocalVersion();
-                                long remoteVersion = blob.getRemoteVersion(blobStore);
-                                if (localVersion != remoteVersion || !blob.isFullyDownloaded()) {
-                                    if (blob.isFullyDownloaded()) {
-                                        //Avoid case of different blob version
-                                        // when blob is not downloaded (first time download)
-                                        numBlobUpdateVersionChanged.mark();
+                                if (blob.isUsed()) {
+                                    long localVersion = blob.getLocalVersion();
+                                    long remoteVersion = blob.getRemoteVersion(blobStore);
+                                    if (localVersion != remoteVersion || !blob.isFullyDownloaded()) {
+                                        if (blob.isFullyDownloaded()) {
+                                            //Avoid case of different blob version
+                                            // when blob is not downloaded (first time download)
+                                            numBlobUpdateVersionChanged.mark();
+                                        }
+                                        Timer.Context t = singleBlobLocalizationDuration.time();
+                                        try {
+                                            long newVersion = blob.fetchUnzipToTemp(blobStore);
+                                            blob.informReferencesAndCommitNewVersion(newVersion);
+                                            t.stop();
+                                        } finally {
+                                            blob.cleanupOrphanedData();
+                                        }
                                     }
-                                    Timer.Context t = singleBlobLocalizationDuration.time();
-                                    try {
-                                        long newVersion = blob.fetchUnzipToTemp(blobStore);
-                                        blob.informReferencesAndCommitNewVersion(newVersion);
-                                        t.stop();
-                                    } finally {
-                                        blob.cleanupOrphanedData();
-                                    }
+                                } else {
+                                    LOG.debug("Skipping update of unused blob {}", blob);
                                 }
                             }
                             done = true;
diff --git a/storm-server/src/main/java/org/apache/storm/pacemaker/PacemakerServer.java b/storm-server/src/main/java/org/apache/storm/pacemaker/PacemakerServer.java
index d8364dc..19e6cbe 100644
--- a/storm-server/src/main/java/org/apache/storm/pacemaker/PacemakerServer.java
+++ b/storm-server/src/main/java/org/apache/storm/pacemaker/PacemakerServer.java
@@ -63,9 +63,8 @@
         switch (auth) {
 
             case "DIGEST":
-                Configuration loginConf = ClientAuthUtils.getConfiguration(config);
                 authMethod = ThriftNettyServerCodec.AuthMethod.DIGEST;
-                this.secret = ClientAuthUtils.makeDigestPayload(loginConf, ClientAuthUtils.LOGIN_CONTEXT_PACEMAKER_DIGEST);
+                this.secret = ClientAuthUtils.makeDigestPayload(config, ClientAuthUtils.LOGIN_CONTEXT_PACEMAKER_DIGEST);
                 if (this.secret == null) {
                     LOG.error("Can't start pacemaker server without digest secret.");
                     throw new RuntimeException("Can't start pacemaker server without digest secret.");
diff --git a/storm-server/src/test/java/org/apache/storm/daemon/supervisor/SlotTest.java b/storm-server/src/test/java/org/apache/storm/daemon/supervisor/SlotTest.java
index 9dfe2d7..1cfea8a 100644
--- a/storm-server/src/test/java/org/apache/storm/daemon/supervisor/SlotTest.java
+++ b/storm-server/src/test/java/org/apache/storm/daemon/supervisor/SlotTest.java
@@ -15,10 +15,13 @@
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.nullValue;
 
+import com.google.common.collect.Maps;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -77,6 +80,14 @@
         return resources;
     }
 
+    static WorkerResources mkWorkerResources(Double cpu, Double mem_on_heap, Double mem_off_heap, Map<String, Double> resources) {
+        WorkerResources workerResources = mkWorkerResources(cpu, mem_on_heap, mem_off_heap);
+        if (resources != null) {
+            workerResources.set_resources(resources);
+        }
+        return workerResources;
+    }
+
     static LSWorkerHeartbeat mkWorkerHB(String id, int port, List<ExecutorInfo> exec, Integer timeSecs) {
         LSWorkerHeartbeat ret = new LSWorkerHeartbeat();
         ret.set_topology_id(id);
@@ -108,12 +119,64 @@
     }
 
     @Test
-    public void testEquivilant() {
+    public void testWorkerResourceEquality() {
+        WorkerResources resourcesRNull = mkWorkerResources(100.0, 100.0, 100.0, null);
+        WorkerResources resourcesREmpty = mkWorkerResources(100.0, 100.0, 100.0, Maps.newHashMap());
+        assertTrue(Slot.customWorkerResourcesEquality(resourcesRNull,resourcesREmpty));
+
+        Map resources = new HashMap<String, Double>();
+        resources.put("network.resource.units", 0.0);
+        WorkerResources resourcesRNetwork = mkWorkerResources(100.0, 100.0, 100.0,resources);
+        assertTrue(Slot.customWorkerResourcesEquality(resourcesREmpty, resourcesRNetwork));
+
+
+        Map resourcesNetwork = new HashMap<String, Double>();
+        resourcesNetwork.put("network.resource.units", 50.0);
+        WorkerResources resourcesRNetworkNonZero = mkWorkerResources(100.0, 100.0, 100.0,resourcesNetwork);
+        assertFalse(Slot.customWorkerResourcesEquality(resourcesREmpty, resourcesRNetworkNonZero));
+
+        Map resourcesNetworkOne = new HashMap<String, Double>();
+        resourcesNetworkOne.put("network.resource.units", 50.0);
+        WorkerResources resourcesRNetworkOne = mkWorkerResources(100.0, 100.0, 100.0,resourcesNetworkOne);
+        assertTrue(Slot.customWorkerResourcesEquality(resourcesRNetworkOne, resourcesRNetworkNonZero));
+
+        Map resourcesNetworkTwo = new HashMap<String, Double>();
+        resourcesNetworkTwo.put("network.resource.units", 100.0);
+        WorkerResources resourcesRNetworkTwo = mkWorkerResources(100.0, 100.0, 100.0,resourcesNetworkTwo);
+        assertFalse(Slot.customWorkerResourcesEquality(resourcesRNetworkOne, resourcesRNetworkTwo));
+
+        WorkerResources resourcesCpuNull = mkWorkerResources(null, 100.0,100.0);
+        WorkerResources resourcesCPUZero = mkWorkerResources(0.0, 100.0,100.0);
+        assertTrue(Slot.customWorkerResourcesEquality(resourcesCpuNull, resourcesCPUZero));
+
+        WorkerResources resourcesOnHeapMemNull = mkWorkerResources(100.0, null,100.0);
+        WorkerResources resourcesOnHeapMemZero = mkWorkerResources(100.0, 0.0,100.0);
+        assertTrue(Slot.customWorkerResourcesEquality(resourcesOnHeapMemNull, resourcesOnHeapMemZero));
+
+        WorkerResources resourcesOffHeapMemNull = mkWorkerResources(100.0, 100.0,null);
+        WorkerResources resourcesOffHeapMemZero = mkWorkerResources(100.0, 100.0,0.0);
+        assertTrue(Slot.customWorkerResourcesEquality(resourcesOffHeapMemNull, resourcesOffHeapMemZero));
+
+    }
+
+    @Test
+    public void testEquivalent() {
         LocalAssignment a = mkLocalAssignment("A", mkExecutorInfoList(1, 2, 3, 4, 5), mkWorkerResources(100.0, 100.0, 100.0));
         LocalAssignment aResized = mkLocalAssignment("A", mkExecutorInfoList(1, 2, 3, 4, 5), mkWorkerResources(100.0, 200.0, 100.0));
         LocalAssignment b = mkLocalAssignment("B", mkExecutorInfoList(1, 2, 3, 4, 5, 6), mkWorkerResources(100.0, 100.0, 100.0));
         LocalAssignment bReordered = mkLocalAssignment("B", mkExecutorInfoList(6, 5, 4, 3, 2, 1), mkWorkerResources(100.0, 100.0, 100.0));
 
+        LocalAssignment c = mkLocalAssignment("C", mkExecutorInfoList(188, 261),mkWorkerResources(400.0,10000.0,0.0));
+
+        WorkerResources workerResources = mkWorkerResources(400.0, 10000.0, 0.0);
+        Map<String, Double> additionalResources = workerResources.get_resources();
+        if( additionalResources == null) additionalResources = new HashMap<>();
+        additionalResources.put("network.resource.units", 0.0);
+
+        workerResources.set_resources(additionalResources);
+        LocalAssignment cReordered = mkLocalAssignment("C", mkExecutorInfoList(188, 261), workerResources);
+
+        assertTrue(Slot.equivalent(c,cReordered));
         assertTrue(Slot.equivalent(null, null));
         assertTrue(Slot.equivalent(a, a));
         assertTrue(Slot.equivalent(b, bReordered));
diff --git a/storm-server/src/test/java/org/apache/storm/localizer/AsyncLocalizerTest.java b/storm-server/src/test/java/org/apache/storm/localizer/AsyncLocalizerTest.java
index d3c1d74..f97c868 100644
--- a/storm-server/src/test/java/org/apache/storm/localizer/AsyncLocalizerTest.java
+++ b/storm-server/src/test/java/org/apache/storm/localizer/AsyncLocalizerTest.java
@@ -126,18 +126,21 @@
             when(jarBlob.getLocalVersion()).thenReturn(-1L);
             when(jarBlob.getRemoteVersion(any())).thenReturn(100L);
             when(jarBlob.fetchUnzipToTemp(any())).thenReturn(100L);
+            when(jarBlob.isUsed()).thenReturn(true);
 
             LocallyCachedTopologyBlob codeBlob = mock(LocallyCachedTopologyBlob.class);
             doReturn(codeBlob).when(victim).getTopoCode(topoId, localAssignment.get_owner());
             when(codeBlob.getLocalVersion()).thenReturn(-1L);
             when(codeBlob.getRemoteVersion(any())).thenReturn(200L);
             when(codeBlob.fetchUnzipToTemp(any())).thenReturn(200L);
+            when(codeBlob.isUsed()).thenReturn(true);
 
             LocallyCachedTopologyBlob confBlob = mock(LocallyCachedTopologyBlob.class);
             doReturn(confBlob).when(victim).getTopoConf(topoId, localAssignment.get_owner());
             when(confBlob.getLocalVersion()).thenReturn(-1L);
             when(confBlob.getRemoteVersion(any())).thenReturn(300L);
             when(confBlob.fetchUnzipToTemp(any())).thenReturn(300L);
+            when(confBlob.isUsed()).thenReturn(true);
 
             when(mockedReflectionUtils.newInstanceImpl(ClientBlobStore.class)).thenReturn(blobStore);
 
diff --git a/storm-server/src/test/java/org/apache/storm/security/auth/AuthTest.java b/storm-server/src/test/java/org/apache/storm/security/auth/AuthTest.java
index 49a4641..7802390 100644
--- a/storm-server/src/test/java/org/apache/storm/security/auth/AuthTest.java
+++ b/storm-server/src/test/java/org/apache/storm/security/auth/AuthTest.java
@@ -576,7 +576,7 @@
     public void getTransportPluginThrowsRunimeTest() {
         Map<String, Object> conf = ConfigUtils.readStormConfig();
         conf.put(Config.STORM_THRIFT_TRANSPORT_PLUGIN, "null.invalid");
-        ClientAuthUtils.getTransportPlugin(ThriftConnectionType.NIMBUS, conf, null);
+        ClientAuthUtils.getTransportPlugin(ThriftConnectionType.NIMBUS, conf);
     }
 
     @Test