diff --git a/lucene/ivy-versions.properties b/lucene/ivy-versions.properties
index 510befa..bc46ee6 100644
--- a/lucene/ivy-versions.properties
+++ b/lucene/ivy-versions.properties
@@ -106,6 +106,7 @@
 org.apache.curator.version = 2.8.0
 /org.apache.curator/curator-client = ${org.apache.curator.version}
 /org.apache.curator/curator-framework = ${org.apache.curator.version}
+/org.apache.curator/curator-recipes = ${org.apache.curator.version}
 
 /org.apache.derby/derby = 10.9.1.0
 
diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index a728c8a..6707e1a 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -100,6 +100,9 @@
 * SOLR-9275: XML QueryParser support (defType=xmlparser) now extensible via configuration.
   (Christine Poerschke)
 
+* SOLR-9200: Add Delegation Token Support to Solr.
+  (Gregory Chanan)
+
 Bug Fixes
 ----------------------
 
diff --git a/solr/core/ivy.xml b/solr/core/ivy.xml
index 5dad49b..08272ad 100644
--- a/solr/core/ivy.xml
+++ b/solr/core/ivy.xml
@@ -134,6 +134,10 @@
     <dependency org="antlr" name="antlr" rev="${/antlr/antlr}" conf="test.MiniKdc"/>
     <dependency org="net.sf.ehcache" name="ehcache-core" rev="${/net.sf.ehcache/ehcache-core}" conf="test.MiniKdc"/>
 
+    <dependency org="org.apache.curator" name="curator-framework" rev="${/org.apache.curator/curator-framework}" conf="compile"/>
+    <dependency org="org.apache.curator" name="curator-client" rev="${/org.apache.curator/curator-client}" conf="compile"/>
+    <dependency org="org.apache.curator" name="curator-recipes" rev="${/org.apache.curator/curator-recipes}" conf="compile"/>
+
     <!-- StatsComponents percentiles Dependencies-->
     <dependency org="com.tdunning" name="t-digest" rev="${/com.tdunning/t-digest}" conf="compile->*"/>
     <!-- SQL Parser -->
diff --git a/solr/core/src/java/org/apache/solr/security/AuthenticationPlugin.java b/solr/core/src/java/org/apache/solr/security/AuthenticationPlugin.java
index 105f307..d8f2ef2 100644
--- a/solr/core/src/java/org/apache/solr/security/AuthenticationPlugin.java
+++ b/solr/core/src/java/org/apache/solr/security/AuthenticationPlugin.java
@@ -17,18 +17,11 @@
 package org.apache.solr.security;
 
 import javax.servlet.FilterChain;
-import javax.servlet.ServletException;
 import javax.servlet.ServletRequest;
 import javax.servlet.ServletResponse;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletRequestWrapper;
 import java.io.Closeable;
-import java.io.IOException;
-import java.security.Principal;
 import java.util.Map;
 
-import org.apache.http.auth.BasicUserPrincipal;
-
 /**
  * 
  * @lucene.experimental
@@ -42,32 +35,20 @@
    * @param pluginConfig Config parameters, possibly from a ZK source
    */
   public abstract void init(Map<String, Object> pluginConfig);
-
-  protected void forward(String user, ServletRequest  req, ServletResponse rsp,
-                                    FilterChain chain) throws IOException, ServletException {
-    if(user != null) {
-      final Principal p = new BasicUserPrincipal(user);
-      req = new HttpServletRequestWrapper((HttpServletRequest) req) {
-        @Override
-        public Principal getUserPrincipal() {
-          return p;
-        }
-      };
-    }
-    chain.doFilter(req,rsp);
-  }
  
   /**
-   * This method must authenticate the request. Upon a successful authentication, this 
+   * This method attempts to authenticate the request. Upon a successful authentication, this
    * must call the next filter in the filter chain and set the user principal of the request,
    * or else, upon an error or an authentication failure, throw an exception.
-   * 
+   *
    * @param request the http request
    * @param response the http response
    * @param filterChain the servlet filter chain
+   * @return false if the request not be processed by Solr (not continue), i.e.
+   * the response and status code have already been sent.
    * @throws Exception any exception thrown during the authentication, e.g. PrivilegedActionException
    */
-  public abstract void doAuthenticate(ServletRequest request, ServletResponse response,
+  public abstract boolean doAuthenticate(ServletRequest request, ServletResponse response,
       FilterChain filterChain) throws Exception;
 
 
diff --git a/solr/core/src/java/org/apache/solr/security/BasicAuthPlugin.java b/solr/core/src/java/org/apache/solr/security/BasicAuthPlugin.java
index 03c75c6..e3f53a2 100644
--- a/solr/core/src/java/org/apache/solr/security/BasicAuthPlugin.java
+++ b/solr/core/src/java/org/apache/solr/security/BasicAuthPlugin.java
@@ -99,7 +99,7 @@
   }
 
   @Override
-  public void doAuthenticate(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain) throws Exception {
+  public boolean doAuthenticate(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain) throws Exception {
 
     HttpServletRequest request = (HttpServletRequest) servletRequest;
     HttpServletResponse response = (HttpServletResponse) servletResponse;
@@ -127,6 +127,7 @@
                   }
                 };
                 filterChain.doFilter(wrapper, response);
+                return true;
               }
 
             } else {
@@ -143,8 +144,10 @@
       } else {
         request.setAttribute(AuthenticationPlugin.class.getName(), zkAuthentication.getPromptHeaders());
         filterChain.doFilter(request, response);
+        return true;
       }
     }
+    return false;
   }
 
   @Override
diff --git a/solr/core/src/java/org/apache/solr/security/DelegationTokenKerberosFilter.java b/solr/core/src/java/org/apache/solr/security/DelegationTokenKerberosFilter.java
new file mode 100644
index 0000000..7dbb1ad
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/security/DelegationTokenKerberosFilter.java
@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.solr.security;
+
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.LinkedList;
+import java.util.List;
+
+import javax.servlet.FilterChain;
+import javax.servlet.FilterConfig;
+import javax.servlet.ServletException;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletRequestWrapper;
+
+import org.apache.curator.RetryPolicy;
+import org.apache.curator.framework.AuthInfo;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.api.ACLProvider;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+
+import org.apache.hadoop.security.authentication.server.AuthenticationHandler;
+import org.apache.hadoop.security.token.delegation.web.DelegationTokenAuthenticationFilter;
+import org.apache.solr.common.cloud.SecurityAwareZkACLProvider;
+import org.apache.solr.common.cloud.SolrZkClient;
+import org.apache.solr.common.cloud.ZkACLProvider;
+import org.apache.solr.common.cloud.ZkCredentialsProvider;
+import org.apache.zookeeper.data.ACL;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DelegationTokenKerberosFilter extends DelegationTokenAuthenticationFilter {
+  private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+  private CuratorFramework curatorFramework;
+
+  @Override
+  public void init(FilterConfig conf) throws ServletException {
+    if (conf != null && "zookeeper".equals(conf.getInitParameter("signer.secret.provider"))) {
+      SolrZkClient zkClient =
+          (SolrZkClient)conf.getServletContext().getAttribute(KerberosPlugin.DELEGATION_TOKEN_ZK_CLIENT);
+      conf.getServletContext().setAttribute("signer.secret.provider.zookeeper.curator.client",
+          getCuratorClient(zkClient));
+    }
+    super.init(conf);
+  }
+
+  @Override
+  public void doFilter(ServletRequest request, ServletResponse response,
+      FilterChain filterChain) throws IOException, ServletException {
+    // HttpClient 4.4.x throws NPE if query string is null and parsed through URLEncodedUtils.
+    // See HTTPCLIENT-1746 and HADOOP-12767
+    HttpServletRequest httpRequest = (HttpServletRequest)request;
+    String queryString = httpRequest.getQueryString();
+    final String nonNullQueryString = queryString == null ? "" : queryString;
+    HttpServletRequest requestNonNullQueryString = new HttpServletRequestWrapper(httpRequest){
+      @Override
+      public String getQueryString() {
+        return nonNullQueryString;
+      }
+    };
+    super.doFilter(requestNonNullQueryString, response, filterChain);
+  }
+
+  @Override
+  public void destroy() {
+    super.destroy();
+    if (curatorFramework != null) curatorFramework.close();
+    curatorFramework = null;
+  }
+
+  @Override
+  protected void initializeAuthHandler(String authHandlerClassName,
+                                       FilterConfig filterConfig) throws ServletException {
+    // set the internal authentication handler in order to record whether the request should continue
+    super.initializeAuthHandler(authHandlerClassName, filterConfig);
+    AuthenticationHandler authHandler = getAuthenticationHandler();
+    super.initializeAuthHandler(KerberosPlugin.RequestContinuesRecorderAuthenticationHandler.class.getName(), filterConfig);
+    KerberosPlugin.RequestContinuesRecorderAuthenticationHandler newAuthHandler =
+        (KerberosPlugin.RequestContinuesRecorderAuthenticationHandler)getAuthenticationHandler();
+    newAuthHandler.setAuthHandler(authHandler);
+  }
+
+  protected CuratorFramework getCuratorClient(SolrZkClient zkClient) {
+    // should we try to build a RetryPolicy off of the ZkController?
+    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
+    if (zkClient == null) {
+      throw new IllegalArgumentException("zkClient required");
+    }
+    String zkHost = zkClient.getZkServerAddress();
+    String zkChroot = zkHost.substring(zkHost.indexOf("/"));
+    zkChroot = zkChroot.startsWith("/") ? zkChroot.substring(1) : zkChroot;
+    String zkNamespace = zkChroot + SecurityAwareZkACLProvider.SECURITY_ZNODE_PATH;
+    String zkConnectionString = zkHost.substring(0, zkHost.indexOf("/"));
+    SolrZkToCuratorCredentialsACLs curatorToSolrZk = new SolrZkToCuratorCredentialsACLs(zkClient);
+    final int connectionTimeoutMs = 30000; // this value is currently hard coded, see SOLR-7561.
+
+    curatorFramework = CuratorFrameworkFactory.builder()
+        .namespace(zkNamespace)
+        .connectString(zkConnectionString)
+        .retryPolicy(retryPolicy)
+        .aclProvider(curatorToSolrZk.getACLProvider())
+        .authorization(curatorToSolrZk.getAuthInfos())
+        .sessionTimeoutMs(zkClient.getZkClientTimeout())
+        .connectionTimeoutMs(connectionTimeoutMs)
+        .build();
+    curatorFramework.start();
+    return curatorFramework;
+  }
+
+  /**
+   * Convert Solr Zk Credentials/ACLs to Curator versions
+   */
+  protected static class SolrZkToCuratorCredentialsACLs {
+    private final ACLProvider aclProvider;
+    private final List<AuthInfo> authInfos;
+
+    public SolrZkToCuratorCredentialsACLs(SolrZkClient zkClient) {
+      this.aclProvider = createACLProvider(zkClient);
+      this.authInfos = createAuthInfo(zkClient);
+    }
+
+    public ACLProvider getACLProvider() { return aclProvider; }
+    public List<AuthInfo> getAuthInfos() { return authInfos; }
+
+    private ACLProvider createACLProvider(SolrZkClient zkClient) {
+      final ZkACLProvider zkACLProvider = zkClient.getZkACLProvider();
+      return new ACLProvider() {
+        @Override
+        public List<ACL> getDefaultAcl() {
+          return zkACLProvider.getACLsToAdd(null);
+        }
+
+        @Override
+        public List<ACL> getAclForPath(String path) {
+           List<ACL> acls = zkACLProvider.getACLsToAdd(path);
+           return acls;
+        }
+      };
+    }
+
+    private List<AuthInfo> createAuthInfo(SolrZkClient zkClient) {
+      List<AuthInfo> ret = new LinkedList<AuthInfo>();
+
+      // In theory the credentials to add could change here if zookeeper hasn't been initialized
+      ZkCredentialsProvider credentialsProvider =
+        zkClient.getZkClientConnectionStrategy().getZkCredentialsToAddAutomatically();
+      for (ZkCredentialsProvider.ZkCredentials zkCredentials : credentialsProvider.getCredentials()) {
+        ret.add(new AuthInfo(zkCredentials.getScheme(), zkCredentials.getAuth()));
+      }
+      return ret;
+    }
+  }
+}
diff --git a/solr/core/src/java/org/apache/solr/security/KerberosFilter.java b/solr/core/src/java/org/apache/solr/security/KerberosFilter.java
index ee23488..9c53050 100644
--- a/solr/core/src/java/org/apache/solr/security/KerberosFilter.java
+++ b/solr/core/src/java/org/apache/solr/security/KerberosFilter.java
@@ -26,6 +26,7 @@
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import org.apache.hadoop.security.authentication.server.AuthenticationFilter;
+import org.apache.hadoop.security.authentication.server.AuthenticationHandler;
 
 public class KerberosFilter extends AuthenticationFilter {
   
@@ -35,6 +36,19 @@
   }
 
   @Override
+  protected void initializeAuthHandler(String authHandlerClassName,
+                                       FilterConfig filterConfig) throws ServletException {
+    // set the internal authentication handler in order to record whether the request should continue
+    super.initializeAuthHandler(authHandlerClassName, filterConfig);
+    AuthenticationHandler authHandler = getAuthenticationHandler();
+    super.initializeAuthHandler(
+        KerberosPlugin.RequestContinuesRecorderAuthenticationHandler.class.getName(), filterConfig);
+    KerberosPlugin.RequestContinuesRecorderAuthenticationHandler newAuthHandler =
+        (KerberosPlugin.RequestContinuesRecorderAuthenticationHandler)getAuthenticationHandler();
+    newAuthHandler.setAuthHandler(authHandler);
+  }
+
+  @Override
   protected void doFilter(FilterChain filterChain, HttpServletRequest request,
       HttpServletResponse response) throws IOException, ServletException {
     super.doFilter(filterChain, request, response);
diff --git a/solr/core/src/java/org/apache/solr/security/KerberosPlugin.java b/solr/core/src/java/org/apache/solr/security/KerberosPlugin.java
index 7a83ab5..1cd476f 100644
--- a/solr/core/src/java/org/apache/solr/security/KerberosPlugin.java
+++ b/solr/core/src/java/org/apache/solr/security/KerberosPlugin.java
@@ -16,14 +16,18 @@
  */
 package org.apache.solr.security;
 
+import java.io.IOException;
 import java.io.InputStream;
+import java.io.PrintWriter;
 import java.lang.invoke.MethodHandles;
 import java.net.MalformedURLException;
 import java.net.URL;
+import java.util.Collections;
 import java.util.Enumeration;
 import java.util.EventListener;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Set;
 
 import javax.servlet.Filter;
@@ -41,12 +45,22 @@
 import javax.servlet.SessionCookieConfig;
 import javax.servlet.SessionTrackingMode;
 import javax.servlet.descriptor.JspConfigDescriptor;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import javax.servlet.http.HttpServletResponseWrapper;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.commons.collections.iterators.IteratorEnumeration;
+import org.apache.hadoop.security.authentication.client.AuthenticationException;
+import org.apache.hadoop.security.authentication.server.AuthenticationHandler;
+import org.apache.hadoop.security.authentication.server.AuthenticationToken;
 import org.apache.solr.client.solrj.impl.Krb5HttpClientBuilder;
 import org.apache.solr.client.solrj.impl.SolrHttpClientBuilder;
+import org.apache.solr.cloud.ZkController;
 import org.apache.solr.common.SolrException;
 import org.apache.solr.common.SolrException.ErrorCode;
+import org.apache.solr.common.cloud.SecurityAwareZkACLProvider;
+import org.apache.solr.common.util.SuppressForbidden;
 import org.apache.solr.core.CoreContainer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -55,7 +69,7 @@
   private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 
   Krb5HttpClientBuilder kerberosBuilder = new Krb5HttpClientBuilder();
-  Filter kerberosFilter = new KerberosFilter();
+  Filter kerberosFilter;
   
   public static final String NAME_RULES_PARAM = "solr.kerberos.name.rules";
   public static final String COOKIE_DOMAIN_PARAM = "solr.kerberos.cookie.domain";
@@ -64,6 +78,26 @@
   public static final String KEYTAB_PARAM = "solr.kerberos.keytab";
   public static final String TOKEN_VALID_PARAM = "solr.kerberos.token.valid";
   public static final String COOKIE_PORT_AWARE_PARAM = "solr.kerberos.cookie.portaware";
+  public static final String DELEGATION_TOKEN_ENABLED = "solr.kerberos.delegation.token.enabled";
+  public static final String DELEGATION_TOKEN_KIND = "solr.kerberos.delegation.token.kind";
+  public static final String DELEGATION_TOKEN_VALIDITY = "solr.kerberos.delegation.token.validity";
+  public static final String DELEGATION_TOKEN_SECRET_PROVIDER = "solr.kerberos.delegation.token.signer.secret.provider";
+  public static final String DELEGATION_TOKEN_SECRET_PROVIDER_ZK_PATH =
+      "solr.kerberos.delegation.token.signer.secret.provider.zookeper.path";
+  public static final String DELEGATION_TOKEN_SECRET_MANAGER_ZNODE_WORKING_PATH =
+      "solr.kerberos.delegation.token.secret.manager.znode.working.path";
+  public static final String DELEGATION_TOKEN_TYPE_DEFAULT = "solr-dt";
+  
+  // filled in by Plugin/Filter
+  static final String REQUEST_CONTINUES_ATTR =
+      "org.apache.solr.security.kerberosplugin.requestcontinues";
+  static final String DELEGATION_TOKEN_ZK_CLIENT =
+      "solr.kerberos.delegation.token.zk.client";
+
+  // allows test to specify an alternate auth handler
+  @VisibleForTesting
+  public static final String AUTH_HANDLER_PARAM = "solr.kerberos.auth.handler";
+
   private final CoreContainer coreContainer;
 
   public KerberosPlugin(CoreContainer coreContainer) {
@@ -74,12 +108,47 @@
   public void init(Map<String, Object> pluginConfig) {
     try {
       Map<String, String> params = new HashMap();
-      params.put("type", "kerberos");
+      putParam(params, "type", AUTH_HANDLER_PARAM, "kerberos");
       putParam(params, "kerberos.name.rules", NAME_RULES_PARAM, "DEFAULT");
       putParam(params, "token.valid", TOKEN_VALID_PARAM, "30");
       putParam(params, "cookie.path", COOKIE_PATH_PARAM, "/");
-      putParam(params, "kerberos.principal", PRINCIPAL_PARAM, null);
-      putParam(params, "kerberos.keytab", KEYTAB_PARAM, null);
+      if ("kerberos".equals(params.get("type"))) {
+        putParam(params, "kerberos.principal", PRINCIPAL_PARAM, null);
+        putParam(params, "kerberos.keytab", KEYTAB_PARAM, null);
+      } else {
+        // allow tests which specify AUTH_HANDLER_PARAM to avoid specifying kerberos principal/keytab
+        putParamOptional(params, "kerberos.principal", PRINCIPAL_PARAM);
+        putParamOptional(params, "kerberos.keytab", KEYTAB_PARAM);
+      }
+
+      String delegationTokenStr = System.getProperty(DELEGATION_TOKEN_ENABLED, null);
+      boolean delegationTokenEnabled =
+          (delegationTokenStr == null) ? false : Boolean.parseBoolean(delegationTokenStr);
+      ZkController controller = coreContainer.getZkController();
+
+      if (delegationTokenEnabled) {
+        putParam(params, "delegation-token.token-kind", DELEGATION_TOKEN_KIND, DELEGATION_TOKEN_TYPE_DEFAULT);
+        if (coreContainer.isZooKeeperAware()) {
+          putParam(params, "signer.secret.provider", DELEGATION_TOKEN_SECRET_PROVIDER, "zookeeper");
+          if ("zookeeper".equals(params.get("signer.secret.provider"))) {
+            String zkHost = controller.getZkServerAddress();
+            putParam(params, "token.validity", DELEGATION_TOKEN_VALIDITY, "36000");
+            params.put("zk-dt-secret-manager.enable", "true");
+            // Note - Curator complains if the znodeWorkingPath starts with /
+            String chrootPath = zkHost.substring(zkHost.indexOf("/"));
+            String relativePath = chrootPath.startsWith("/") ? chrootPath.substring(1) : chrootPath;
+            putParam(params, "zk-dt-secret-manager.znodeWorkingPath",
+                DELEGATION_TOKEN_SECRET_MANAGER_ZNODE_WORKING_PATH,
+                relativePath + SecurityAwareZkACLProvider.SECURITY_ZNODE_PATH + "/zkdtsm");
+            putParam(params, "signer.secret.provider.zookeeper.path",
+                DELEGATION_TOKEN_SECRET_PROVIDER_ZK_PATH, "/token");
+            // ensure krb5 is setup properly before running curator
+            getHttpClientBuilder(SolrHttpClientBuilder.create());
+          }
+        } else {
+          log.info("CoreContainer is not ZooKeeperAware, not setting ZK-related delegation token properties");
+        }
+      }
 
       // Special handling for the "cookie.domain" based on whether port should be
       // appended to the domain. Useful for situations where multiple solr nodes are
@@ -94,16 +163,27 @@
         if (host==null) {
           throw new SolrException(ErrorCode.SERVER_ERROR, "Missing required parameter '"+COOKIE_DOMAIN_PARAM+"'.");
         }
-        int port = coreContainer.getZkController().getHostPort();
+        int port = controller.getHostPort();
         params.put("cookie.domain", host + ":" + port);
       }
-      
+
+      final ServletContext servletContext = new AttributeOnlyServletContext();
+      if (delegationTokenEnabled) {
+        kerberosFilter = new DelegationTokenKerberosFilter();
+        // pass an attribute-enabled context in order to pass the zkClient
+        // and because the filter may pass a curator instance.
+        if (controller != null) {
+          servletContext.setAttribute(DELEGATION_TOKEN_ZK_CLIENT, controller.getZkClient());
+        }
+      } else {
+        kerberosFilter = new KerberosFilter();
+      }
       log.info("Params: "+params);
 
       FilterConfig conf = new FilterConfig() {
         @Override
         public ServletContext getServletContext() {
-          return noContext;
+          return servletContext;
         }
 
         @Override
@@ -136,11 +216,43 @@
     params.put(internalParamName, value);
   }
 
+  private void putParamOptional(Map<String, String> params, String internalParamName, String externalParamName) {
+    String value = System.getProperty(externalParamName);
+    if (value!=null) {
+      params.put(internalParamName, value);
+    }
+  }
+
   @Override
-  public void doAuthenticate(ServletRequest req, ServletResponse rsp,
+  public boolean doAuthenticate(ServletRequest req, ServletResponse rsp,
       FilterChain chain) throws Exception {
     log.debug("Request to authenticate using kerberos: "+req);
-    kerberosFilter.doFilter(req, rsp, chain);    
+
+    final HttpServletResponse frsp = (HttpServletResponse)rsp;
+
+    // kerberosFilter may close the stream and write to closed streams,
+    // see HADOOP-13346.  To work around, pass a PrintWriter that ignores
+    // closes
+    HttpServletResponse rspCloseShield = new HttpServletResponseWrapper(frsp) {
+      @SuppressForbidden(reason = "Hadoop DelegationTokenAuthenticationFilter uses response writer, this" +
+          "is providing a CloseShield on top of that")
+      @Override
+      public PrintWriter getWriter() throws IOException {
+        final PrintWriter pw = new PrintWriterWrapper(frsp.getWriter()) {
+          @Override
+          public void close() {};
+        };
+        return pw;
+      }
+    };
+    kerberosFilter.doFilter(req, rspCloseShield, chain);
+    String requestContinuesAttr = (String)req.getAttribute(REQUEST_CONTINUES_ATTR);
+    if (requestContinuesAttr == null) {
+      log.warn("Could not find " + REQUEST_CONTINUES_ATTR);
+      return false;
+    } else {
+      return Boolean.parseBoolean(requestContinuesAttr);
+    }
   }
 
   @Override
@@ -153,8 +265,9 @@
     kerberosBuilder.close();
   }
 
-  protected static ServletContext noContext = new ServletContext() {
-    
+  protected static class AttributeOnlyServletContext implements ServletContext {
+    private Map<String, Object> attributes = new HashMap<String, Object>();
+
     @Override
     public void setSessionTrackingModes(Set<SessionTrackingMode> sessionTrackingModes) {}
     
@@ -162,12 +275,16 @@
     public boolean setInitParameter(String name, String value) {
       return false;
     }
-    
+
     @Override
-    public void setAttribute(String name, Object object) {}
-    
+    public void setAttribute(String name, Object object) {
+      attributes.put(name, object);
+    }
+
     @Override
-    public void removeAttribute(String name) {}
+    public void removeAttribute(String name) {
+      attributes.remove(name);
+    }
     
     @Override
     public void log(String message, Throwable throwable) {}
@@ -327,15 +444,15 @@
     public ClassLoader getClassLoader() {
       return null;
     }
-    
+
     @Override
     public Enumeration<String> getAttributeNames() {
-      return null;
+      return Collections.enumeration(attributes.keySet());
     }
-    
+
     @Override
     public Object getAttribute(String name) {
-      return null;
+      return attributes.get(name);
     }
     
     @Override
@@ -395,4 +512,44 @@
       return null;
     }
   };
+
+  /*
+   * {@link AuthenticationHandler} that delegates to another {@link AuthenticationHandler}
+   * and records the response of managementOperation (which indicates whether the request
+   * should continue or not).
+   */
+  public static class RequestContinuesRecorderAuthenticationHandler implements AuthenticationHandler {
+    private AuthenticationHandler authHandler;
+
+    public void setAuthHandler(AuthenticationHandler authHandler) {
+      this.authHandler = authHandler;
+    }
+
+    public String getType() {
+      return authHandler.getType();
+    }
+
+    public void init(Properties config) throws ServletException {
+      // authHandler has already been init'ed, nothing to do here
+    }
+
+    public void destroy() {
+      authHandler.destroy();
+    }
+
+    public boolean managementOperation(AuthenticationToken token,
+                                       HttpServletRequest request,
+                                       HttpServletResponse response)
+        throws IOException, AuthenticationException {
+      boolean result = authHandler.managementOperation(token, request, response);
+      request.setAttribute(KerberosPlugin.REQUEST_CONTINUES_ATTR, new Boolean(result).toString());
+      return result;
+    }
+
+
+    public AuthenticationToken authenticate(HttpServletRequest request, HttpServletResponse response)
+        throws IOException, AuthenticationException {
+      return authHandler.authenticate(request, response);
+    }
+  }
 }
diff --git a/solr/core/src/java/org/apache/solr/security/PKIAuthenticationPlugin.java b/solr/core/src/java/org/apache/solr/security/PKIAuthenticationPlugin.java
index fb3665c..e5d2653 100644
--- a/solr/core/src/java/org/apache/solr/security/PKIAuthenticationPlugin.java
+++ b/solr/core/src/java/org/apache/solr/security/PKIAuthenticationPlugin.java
@@ -89,12 +89,12 @@
 
   @SuppressForbidden(reason = "Needs currentTimeMillis to compare against time in header")
   @Override
-  public void doAuthenticate(ServletRequest request, ServletResponse response, FilterChain filterChain) throws Exception {
+  public boolean doAuthenticate(ServletRequest request, ServletResponse response, FilterChain filterChain) throws Exception {
 
     String requestURI = ((HttpServletRequest) request).getRequestURI();
     if (requestURI.endsWith(PATH)) {
       filterChain.doFilter(request, response);
-      return;
+      return true;
     }
     long receivedTime = System.currentTimeMillis();
     String header = ((HttpServletRequest) request).getHeader(HEADER);
@@ -102,14 +102,14 @@
       //this must not happen
       log.error("No SolrAuth header present");
       filterChain.doFilter(request, response);
-      return;
+      return true;
     }
 
     List<String> authInfo = StrUtils.splitWS(header, false);
     if (authInfo.size() < 2) {
       log.error("Invalid SolrAuth Header {}", header);
       filterChain.doFilter(request, response);
-      return;
+      return true;
     }
 
     String nodeName = authInfo.get(0);
@@ -119,12 +119,12 @@
     if (decipher == null) {
       log.error("Could not decipher a header {} . No principal set", header);
       filterChain.doFilter(request, response);
-      return;
+      return true;
     }
     if ((receivedTime - decipher.timestamp) > MAX_VALIDITY) {
       log.error("Invalid key request timestamp: {} , received timestamp: {} , TTL: {}", decipher.timestamp, receivedTime, MAX_VALIDITY);
         filterChain.doFilter(request, response);
-        return;
+        return true;
     }
 
     final Principal principal = "$".equals(decipher.userName) ?
@@ -132,6 +132,7 @@
         new BasicUserPrincipal(decipher.userName);
 
     filterChain.doFilter(getWrapper((HttpServletRequest) request, principal), response);
+    return true;
   }
 
   private static HttpServletRequestWrapper getWrapper(final HttpServletRequest request, final Principal principal) {
diff --git a/solr/core/src/java/org/apache/solr/security/PrintWriterWrapper.java b/solr/core/src/java/org/apache/solr/security/PrintWriterWrapper.java
new file mode 100644
index 0000000..a4e47b5
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/security/PrintWriterWrapper.java
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.solr.security;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.Locale;
+
+import org.apache.commons.lang.NotImplementedException;
+
+/**
+ * Wrapper for PrintWriter that delegates to constructor arg
+ */
+public class PrintWriterWrapper extends PrintWriter {
+  private PrintWriter printWriter;
+
+  public PrintWriterWrapper(PrintWriter printWriter) {
+    super(new StringWriter());
+    this.printWriter = printWriter;
+  }
+
+  @Override
+  public PrintWriter append(char c) {
+    return printWriter.append(c);
+  }
+
+  @Override
+  public PrintWriter append(CharSequence csq) {
+    return printWriter.append(csq);
+  }
+
+  @Override
+  public PrintWriter append(CharSequence csq, int start, int end) {
+    return printWriter.append(csq, start, end);
+  }
+
+  @Override
+  public boolean checkError() {
+    return printWriter.checkError();
+  }
+
+  @Override
+  protected void clearError() {
+    throw new NotImplementedException();
+  }
+
+  @Override
+  public void close() {
+    printWriter.close();
+  }
+
+  @Override
+  public void flush() {
+    printWriter.flush();
+  }
+
+  @Override
+  public PrintWriter format(Locale l, String format, Object... args) {
+    return printWriter.format(l, format, args);
+  }
+
+  @Override
+  public PrintWriter format(String format, Object... args) {
+    throw new NotImplementedException("Forbidden API");
+  }
+
+  @Override
+  public void print(boolean b) {
+    printWriter.print(b);
+  }
+
+  @Override
+  public void print(char c) {
+    printWriter.print(c);
+  }
+
+  @Override
+  public void print(char[] s) {
+    printWriter.print(s);
+  }
+
+  @Override
+  public void print(double d) {
+    printWriter.print(d);
+  }
+
+  @Override
+  public void print(float f) {
+    printWriter.print(f);
+  }
+
+  @Override
+  public void print(int i) {
+    printWriter.print(i);
+  }
+
+  @Override
+  public void print(long l) {
+    printWriter.print(l);
+  }
+
+  @Override
+  public void print(Object obj) {
+    printWriter.print(obj);
+  }
+
+  @Override
+  public void print(String s) {
+    printWriter.print(s);
+  }
+
+  @Override
+  public PrintWriter printf(Locale l, String format, Object... args) {
+    return printWriter.printf(l, format, args);
+  }
+
+  @Override
+  public PrintWriter printf(String format, Object... args) {
+    throw new NotImplementedException("Forbidden API");
+  }
+
+  @Override
+  public void println() {
+    printWriter.println();
+  }
+
+  @Override
+  public void println(boolean x) {
+    printWriter.println(x);
+  }
+
+  @Override
+  public void println(char x) {
+    printWriter.println(x);
+  }
+
+  @Override
+  public void println(char[] x) {
+    printWriter.println(x);
+  }
+
+  @Override
+  public void println(double x) {
+    printWriter.println(x);
+  }
+
+  @Override
+  public void println(float x) {
+    printWriter.println(x);
+  }
+
+  @Override
+  public void println(int x) {
+    printWriter.println(x);
+  }
+
+  @Override
+  public void println(long x) {
+    printWriter.println(x);
+  }
+
+  @Override
+  public void println(Object x) {
+    printWriter.println(x);
+  }
+
+  @Override
+  public void println(String x) {
+    printWriter.println(x);
+  }
+
+  @Override
+  protected void setError() {
+    throw new NotImplementedException();
+  }
+
+  @Override
+  public void write(char[] buf) {
+    printWriter.write(buf);
+  }
+
+  @Override
+  public void write(char[] buf, int off, int len) {
+    printWriter.write(buf, off, len);
+  }
+
+  @Override
+  public void write(int c) {
+    printWriter.write(c);
+  }
+
+  @Override
+  public void write(String s) {
+    printWriter.write(s);
+  }
+
+  @Override
+  public void write(String s, int off, int len) {
+    printWriter.write(s, off, len);
+  }
+}
diff --git a/solr/core/src/java/org/apache/solr/servlet/SolrDispatchFilter.java b/solr/core/src/java/org/apache/solr/servlet/SolrDispatchFilter.java
index 2d08935..4a680e5 100644
--- a/solr/core/src/java/org/apache/solr/servlet/SolrDispatchFilter.java
+++ b/solr/core/src/java/org/apache/solr/servlet/SolrDispatchFilter.java
@@ -296,6 +296,7 @@
   }
 
   private boolean authenticateRequest(ServletRequest request, ServletResponse response, final AtomicReference<ServletRequest> wrappedRequest) throws IOException {
+    boolean requestContinues = false;
     final AtomicBoolean isAuthenticated = new AtomicBoolean(false);
     AuthenticationPlugin authenticationPlugin = cores.getAuthenticationPlugin();
     if (authenticationPlugin == null) {
@@ -308,7 +309,7 @@
       try {
         log.debug("Request to authenticate: {}, domain: {}, port: {}", request, request.getLocalName(), request.getLocalPort());
         // upon successful authentication, this should call the chain's next filter.
-        authenticationPlugin.doAuthenticate(request, response, new FilterChain() {
+        requestContinues = authenticationPlugin.doAuthenticate(request, response, new FilterChain() {
           public void doFilter(ServletRequest req, ServletResponse rsp) throws IOException, ServletException {
             isAuthenticated.set(true);
             wrappedRequest.set(req);
@@ -319,8 +320,13 @@
         throw new SolrException(ErrorCode.SERVER_ERROR, "Error during request authentication, ", e);
       }
     }
-    // failed authentication?
-    if (!isAuthenticated.get()) {
+    // requestContinues is an optional short circuit, thus we still need to check isAuthenticated.
+    // This is because the AuthenticationPlugin doesn't always have enough information to determine if
+    // it should short circuit, e.g. the Kerberos Authentication Filter will send an error and not
+    // call later filters in chain, but doesn't throw an exception.  We could force each Plugin
+    // to implement isAuthenticated to simplify the check here, but that just moves the complexity to
+    // multiple code paths.
+    if (!requestContinues || !isAuthenticated.get()) {
       response.flushBuffer();
       return false;
     }
diff --git a/solr/core/src/test/org/apache/solr/cloud/KerberosTestServices.java b/solr/core/src/test/org/apache/solr/cloud/KerberosTestServices.java
new file mode 100644
index 0000000..eb27218
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/KerberosTestServices.java
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.solr.cloud;
+
+import javax.security.auth.login.AppConfigurationEntry;
+import javax.security.auth.login.Configuration;
+import java.io.File;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Properties;
+
+import com.carrotsearch.ant.tasks.junit4.dependencies.com.google.common.base.Preconditions;
+import org.apache.hadoop.minikdc.MiniKdc;
+import org.apache.solr.client.solrj.impl.Krb5HttpClientBuilder;
+
+public class KerberosTestServices {
+
+  private MiniKdc kdc;
+  private JaasConfiguration jaasConfiguration;
+  private Configuration savedConfig;
+  private Locale savedLocale;
+
+  private KerberosTestServices(MiniKdc kdc,
+                               JaasConfiguration jaasConfiguration,
+                               Configuration savedConfig,
+                               Locale savedLocale) {
+    this.kdc = kdc;
+    this.jaasConfiguration = jaasConfiguration;
+    this.savedConfig = savedConfig;
+    this.savedLocale = savedLocale;
+  }
+
+  public MiniKdc getKdc() {
+    return kdc;
+  }
+
+  public void start() throws Exception {
+    if (kdc != null) kdc.start();
+    Configuration.setConfiguration(jaasConfiguration);
+    Krb5HttpClientBuilder.regenerateJaasConfiguration();
+    if (brokenLanguagesWithMiniKdc.contains(Locale.getDefault().getLanguage())) {
+      Locale.setDefault(Locale.US);
+    }
+  }
+
+  public void stop() {
+    if (kdc != null) kdc.stop();
+    Configuration.setConfiguration(savedConfig);
+    Krb5HttpClientBuilder.regenerateJaasConfiguration();
+    Locale.setDefault(savedLocale);
+  }
+
+  public static Builder builder() {
+    return new Builder();
+  }
+
+  /**
+   * Returns a MiniKdc that can be used for creating kerberos principals
+   * and keytabs.  Caller is responsible for starting/stopping the kdc.
+   */
+  private static MiniKdc getKdc(File workDir) throws Exception {
+    Properties conf = MiniKdc.createConf();
+    return new MiniKdc(conf, workDir);
+  }
+
+  /**
+   * Programmatic version of a jaas.conf file suitable for connecting
+   * to a SASL-configured zookeeper.
+   */
+  private static class JaasConfiguration extends Configuration {
+
+    private static AppConfigurationEntry[] clientEntry;
+    private static AppConfigurationEntry[] serverEntry;
+    private String clientAppName = "Client", serverAppName = "Server";
+
+    /**
+     * Add an entry to the jaas configuration with the passed in name,
+     * principal, and keytab. The other necessary options will be set for you.
+     *
+     * @param clientPrincipal The principal of the client
+     * @param clientKeytab The location of the keytab with the clientPrincipal
+     * @param serverPrincipal The principal of the server
+     * @param serverKeytab The location of the keytab with the serverPrincipal
+     */
+    public JaasConfiguration(String clientPrincipal, File clientKeytab,
+                             String serverPrincipal, File serverKeytab) {
+      Map<String, String> clientOptions = new HashMap();
+      clientOptions.put("principal", clientPrincipal);
+      clientOptions.put("keyTab", clientKeytab.getAbsolutePath());
+      clientOptions.put("useKeyTab", "true");
+      clientOptions.put("storeKey", "true");
+      clientOptions.put("useTicketCache", "false");
+      clientOptions.put("refreshKrb5Config", "true");
+      String jaasProp = System.getProperty("solr.jaas.debug");
+      if (jaasProp != null && "true".equalsIgnoreCase(jaasProp)) {
+        clientOptions.put("debug", "true");
+      }
+      clientEntry = new AppConfigurationEntry[]{
+          new AppConfigurationEntry(getKrb5LoginModuleName(),
+              AppConfigurationEntry.LoginModuleControlFlag.REQUIRED,
+              clientOptions)};
+      if(serverPrincipal!=null && serverKeytab!=null) {
+        Map<String, String> serverOptions = new HashMap(clientOptions);
+        serverOptions.put("principal", serverPrincipal);
+        serverOptions.put("keytab", serverKeytab.getAbsolutePath());
+        serverEntry =  new AppConfigurationEntry[]{
+            new AppConfigurationEntry(getKrb5LoginModuleName(),
+                AppConfigurationEntry.LoginModuleControlFlag.REQUIRED,
+                serverOptions)};
+      }
+    }
+
+    /**
+     * Add an entry to the jaas configuration with the passed in principal and keytab,
+     * along with the app name.
+     *
+     * @param principal The principal
+     * @param keytab The keytab containing credentials for the principal
+     * @param appName The app name of the configuration
+     */
+    public JaasConfiguration(String principal, File keytab, String appName) {
+      this(principal, keytab, null, null);
+      clientAppName = appName;
+      serverAppName = null;
+    }
+
+    @Override
+    public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
+      if (name.equals(clientAppName)) {
+        return clientEntry;
+      } else if (name.equals(serverAppName)) {
+        return serverEntry;
+      }
+      return null;
+    }
+
+    private String getKrb5LoginModuleName() {
+      String krb5LoginModuleName;
+      if (System.getProperty("java.vendor").contains("IBM")) {
+        krb5LoginModuleName = "com.ibm.security.auth.module.Krb5LoginModule";
+      } else {
+        krb5LoginModuleName = "com.sun.security.auth.module.Krb5LoginModule";
+      }
+      return krb5LoginModuleName;
+    }
+  }
+
+  /**
+   *  These Locales don't generate dates that are compatibile with Hadoop MiniKdc.
+   */
+  private final static List<String> brokenLanguagesWithMiniKdc =
+      Arrays.asList(
+          new Locale("th").getLanguage(),
+          new Locale("ja").getLanguage(),
+          new Locale("hi").getLanguage()
+      );
+
+  public static class Builder {
+    private File kdcWorkDir;
+    private String clientPrincipal;
+    private File clientKeytab;
+    private String serverPrincipal;
+    private File serverKeytab;
+    private String appName;
+    private Locale savedLocale;
+
+    public Builder() {
+      savedLocale = Locale.getDefault();
+    }
+
+    public Builder withKdc(File kdcWorkDir) {
+      this.kdcWorkDir = kdcWorkDir;
+      return this;
+    }
+
+    public Builder withJaasConfiguration(String clientPrincipal, File clientKeytab,
+                                         String serverPrincipal, File serverKeytab) {
+      Preconditions.checkNotNull(clientPrincipal);
+      Preconditions.checkNotNull(clientKeytab);
+      this.clientPrincipal = clientPrincipal;
+      this.clientKeytab = clientKeytab;
+      this.serverPrincipal = serverPrincipal;
+      this.serverKeytab = serverKeytab;
+      this.appName = null;
+      return this;
+    }
+
+    public Builder withJaasConfiguration(String principal, File keytab, String appName) {
+      Preconditions.checkNotNull(principal);
+      Preconditions.checkNotNull(keytab);
+      this.clientPrincipal = principal;
+      this.clientKeytab = keytab;
+      this.serverPrincipal = null;
+      this.serverKeytab = null;
+      this.appName = appName;
+      return this;
+    }
+
+    public KerberosTestServices build() throws Exception {
+      final MiniKdc kdc = kdcWorkDir != null ? getKdc(kdcWorkDir) : null;
+      final Configuration oldConfig = clientPrincipal != null ? Configuration.getConfiguration() : null;
+      JaasConfiguration jaasConfiguration = null;
+      if (clientPrincipal != null) {
+        jaasConfiguration = (appName == null) ?
+            new JaasConfiguration(clientPrincipal, clientKeytab, serverPrincipal, serverKeytab) :
+            new JaasConfiguration(clientPrincipal, clientKeytab, appName);
+      }
+      return new KerberosTestServices(kdc, jaasConfiguration, oldConfig, savedLocale);
+    }
+  }
+}
diff --git a/solr/core/src/test/org/apache/solr/cloud/KerberosTestUtil.java b/solr/core/src/test/org/apache/solr/cloud/KerberosTestUtil.java
deleted file mode 100644
index 7f544ef..0000000
--- a/solr/core/src/test/org/apache/solr/cloud/KerberosTestUtil.java
+++ /dev/null
@@ -1,147 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.solr.cloud;
-
-import java.io.File;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Properties;
-
-import javax.security.auth.login.AppConfigurationEntry;
-import javax.security.auth.login.Configuration;
-
-import org.apache.hadoop.minikdc.MiniKdc;
-
-public class KerberosTestUtil {
-
-  /**
-   * Returns a MiniKdc that can be used for creating kerberos principals
-   * and keytabs.  Caller is responsible for starting/stopping the kdc.
-   */
-  public static MiniKdc getKdc(File workDir) throws Exception {
-    Properties conf = MiniKdc.createConf();
-    return new MiniKdc(conf, workDir);
-  }
-
-  /**
-   * Programmatic version of a jaas.conf file suitable for connecting
-   * to a SASL-configured zookeeper.
-   */
-  public static class JaasConfiguration extends Configuration {
-
-    private static AppConfigurationEntry[] clientEntry;
-    private static AppConfigurationEntry[] serverEntry;
-    private String clientAppName = "Client", serverAppName = "Server";
-
-    /**
-     * Add an entry to the jaas configuration with the passed in name,
-     * principal, and keytab. The other necessary options will be set for you.
-     *
-     * @param clientPrincipal The principal of the client
-     * @param clientKeytab The location of the keytab with the clientPrincipal
-     * @param serverPrincipal The principal of the server
-     * @param serverKeytab The location of the keytab with the serverPrincipal
-     */
-    public JaasConfiguration(String clientPrincipal, File clientKeytab,
-        String serverPrincipal, File serverKeytab) {
-      Map<String, String> clientOptions = new HashMap();
-      clientOptions.put("principal", clientPrincipal);
-      clientOptions.put("keyTab", clientKeytab.getAbsolutePath());
-      clientOptions.put("useKeyTab", "true");
-      clientOptions.put("storeKey", "true");
-      clientOptions.put("useTicketCache", "false");
-      clientOptions.put("refreshKrb5Config", "true");
-      String jaasProp = System.getProperty("solr.jaas.debug");
-      if (jaasProp != null && "true".equalsIgnoreCase(jaasProp)) {
-        clientOptions.put("debug", "true");
-      }
-      clientEntry = new AppConfigurationEntry[]{
-        new AppConfigurationEntry(getKrb5LoginModuleName(),
-        AppConfigurationEntry.LoginModuleControlFlag.REQUIRED,
-        clientOptions)};
-      if(serverPrincipal!=null && serverKeytab!=null) {
-        Map<String, String> serverOptions = new HashMap(clientOptions);
-        serverOptions.put("principal", serverPrincipal);
-        serverOptions.put("keytab", serverKeytab.getAbsolutePath());
-        serverEntry =  new AppConfigurationEntry[]{
-            new AppConfigurationEntry(getKrb5LoginModuleName(),
-                AppConfigurationEntry.LoginModuleControlFlag.REQUIRED,
-                serverOptions)};
-      }
-    }
-
-    /**
-     * Add an entry to the jaas configuration with the passed in principal and keytab, 
-     * along with the app name.
-     * 
-     * @param principal The principal
-     * @param keytab The keytab containing credentials for the principal
-     * @param appName The app name of the configuration
-     */
-    public JaasConfiguration(String principal, File keytab, String appName) {
-      this(principal, keytab, null, null);
-      clientAppName = appName;
-      serverAppName = null;
-    }
-
-    @Override
-    public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
-      if (name.equals(clientAppName)) {
-        return clientEntry;
-      } else if (name.equals(serverAppName)) {
-        return serverEntry;
-      }
-      return null;
-    }
-
-    private String getKrb5LoginModuleName() {
-      String krb5LoginModuleName;
-      if (System.getProperty("java.vendor").contains("IBM")) {
-        krb5LoginModuleName = "com.ibm.security.auth.module.Krb5LoginModule";
-      } else {
-        krb5LoginModuleName = "com.sun.security.auth.module.Krb5LoginModule";
-      }
-      return krb5LoginModuleName;
-    }
-  }
-
-  /**
-   *  These Locales don't generate dates that are compatibile with Hadoop MiniKdc.
-   */
-  private final static List<String> brokenLanguagesWithMiniKdc =
-      Arrays.asList(
-          new Locale("th").getLanguage(), 
-          new Locale("ja").getLanguage(), 
-          new Locale("hi").getLanguage()
-          );
-  /** 
-   *returns the currently set locale, and overrides it with {@link Locale#US} if it's 
-   * currently something MiniKdc can not handle
-   *
-   * @see Locale#setDefault
-   */
-  public static final Locale overrideLocaleIfNotSpportedByMiniKdc() {
-    Locale old = Locale.getDefault();
-    if (brokenLanguagesWithMiniKdc.contains(Locale.getDefault().getLanguage())) {
-      Locale.setDefault(Locale.US);
-    }
-    return old;
-  }
-}
diff --git a/solr/core/src/test/org/apache/solr/cloud/OutOfBoxZkACLAndCredentialsProvidersTest.java b/solr/core/src/test/org/apache/solr/cloud/OutOfBoxZkACLAndCredentialsProvidersTest.java
index 51ad523..0884576 100644
--- a/solr/core/src/test/org/apache/solr/cloud/OutOfBoxZkACLAndCredentialsProvidersTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/OutOfBoxZkACLAndCredentialsProvidersTest.java
@@ -23,6 +23,7 @@
 import java.util.List;
 
 import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.common.cloud.SecurityAwareZkACLProvider;
 import org.apache.solr.common.cloud.SolrZkClient;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.ZooDefs;
@@ -77,6 +78,7 @@
     zkClient.makePath("/protectedMakePathNode", "content".getBytes(DATA_ENCODING), CreateMode.PERSISTENT, false);
     zkClient.create("/unprotectedCreateNode", "content".getBytes(DATA_ENCODING), CreateMode.PERSISTENT, false);
     zkClient.makePath("/unprotectedMakePathNode", "content".getBytes(DATA_ENCODING), CreateMode.PERSISTENT, false);
+    zkClient.create(SecurityAwareZkACLProvider.SECURITY_ZNODE_PATH, "content".getBytes(DATA_ENCODING), CreateMode.PERSISTENT, false);
     zkClient.close();
 
     log.info("####SETUP_END " + getTestName());
@@ -93,7 +95,9 @@
   public void testOutOfBoxSolrZkClient() throws Exception {
     SolrZkClient zkClient = new SolrZkClient(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     try {
-      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient, true, true, true, true, true);
+      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient,
+          true, true, true, true, true,
+          true, true, true, true, true);
     } finally {
       zkClient.close();
     }
@@ -110,6 +114,7 @@
       assertTrue(verifiedList.contains("/solr/unprotectedMakePathNode"));
       assertTrue(verifiedList.contains("/solr/protectedMakePathNode"));
       assertTrue(verifiedList.contains("/solr/protectedCreateNode"));
+      assertTrue(verifiedList.contains("/solr" + SecurityAwareZkACLProvider.SECURITY_ZNODE_PATH));
     } finally {
       zkClient.close();
     }
diff --git a/solr/core/src/test/org/apache/solr/cloud/OverriddenZkACLAndCredentialsProvidersTest.java b/solr/core/src/test/org/apache/solr/cloud/OverriddenZkACLAndCredentialsProvidersTest.java
index b87ab1b..56c0df9 100644
--- a/solr/core/src/test/org/apache/solr/cloud/OverriddenZkACLAndCredentialsProvidersTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/OverriddenZkACLAndCredentialsProvidersTest.java
@@ -18,18 +18,15 @@
 
 import org.apache.solr.SolrTestCaseJ4;
 import org.apache.solr.common.StringUtils;
-import org.apache.solr.common.cloud.DefaultZkACLProvider;
 import org.apache.solr.common.cloud.DefaultZkCredentialsProvider;
+import org.apache.solr.common.cloud.SecurityAwareZkACLProvider;
 import org.apache.solr.common.cloud.SolrZkClient;
 import org.apache.solr.common.cloud.VMParamsAllAndReadonlyDigestZkACLProvider;
 import org.apache.solr.common.cloud.VMParamsSingleSetCredentialsDigestZkCredentialsProvider;
 import org.apache.solr.common.cloud.ZkACLProvider;
 import org.apache.solr.common.cloud.ZkCredentialsProvider;
 import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.data.ACL;
-import org.apache.zookeeper.data.Id;
-import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
@@ -40,7 +37,6 @@
 import java.io.UnsupportedEncodingException;
 import java.lang.invoke.MethodHandles;
 import java.nio.charset.Charset;
-import java.security.NoSuchAlgorithmException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -88,6 +84,7 @@
         "readonlyACLUsername", "readonlyACLPassword").getSolrZkClient(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     zkClient.create("/protectedCreateNode", "content".getBytes(DATA_ENCODING), CreateMode.PERSISTENT, false);
     zkClient.makePath("/protectedMakePathNode", "content".getBytes(DATA_ENCODING), CreateMode.PERSISTENT, false);
+    zkClient.create(SecurityAwareZkACLProvider.SECURITY_ZNODE_PATH, "content".getBytes(DATA_ENCODING), CreateMode.PERSISTENT, false);
     zkClient.close();
     
     zkClient = new SolrZkClientFactoryUsingCompletelyNewProviders(null, null, 
@@ -114,7 +111,9 @@
     SolrZkClient zkClient = new SolrZkClientFactoryUsingCompletelyNewProviders(null, null, 
         null, null).getSolrZkClient(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     try {
-      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient, false, false, false, false, false);
+      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient,
+          false, false, false, false, false,
+          false, false, false, false, false);
     } finally {
       zkClient.close();
     }
@@ -125,7 +124,9 @@
     SolrZkClient zkClient = new SolrZkClientFactoryUsingCompletelyNewProviders("connectAndAllACLUsername", "connectAndAllACLPasswordWrong", 
         null, null).getSolrZkClient(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     try {
-      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient, false, false, false, false, false);
+      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient,
+          false, false, false, false, false,
+          false, false, false, false, false);
     } finally {
       zkClient.close();
     }
@@ -136,7 +137,9 @@
     SolrZkClient zkClient = new SolrZkClientFactoryUsingCompletelyNewProviders("connectAndAllACLUsername", "connectAndAllACLPassword", 
         null, null).getSolrZkClient(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     try {
-      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient, true, true, true, true, true);
+      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient,
+          true, true, true, true, true,
+          true, true, true, true, true);
     } finally {
       zkClient.close();
     }
@@ -147,7 +150,9 @@
     SolrZkClient zkClient = new SolrZkClientFactoryUsingCompletelyNewProviders("readonlyACLUsername", "readonlyACLPassword",
         null, null).getSolrZkClient(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     try {
-      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient, true, true, false, false, false);
+      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient,
+          true, true, false, false, false,
+          false, false, false, false, false);
     } finally {
       zkClient.close();
     }
@@ -159,7 +164,9 @@
     
     SolrZkClient zkClient = new SolrZkClientUsingVMParamsProvidersButWithDifferentVMParamsNames(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     try {
-      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient, false, false, false, false, false);
+      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient,
+          false, false, false, false, false,
+          false, false, false, false, false);
     } finally {
       zkClient.close();
     }
@@ -171,7 +178,9 @@
     
     SolrZkClient zkClient = new SolrZkClientUsingVMParamsProvidersButWithDifferentVMParamsNames(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     try {
-      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient, false, false, false, false, false);
+      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient,
+          false, false, false, false, false,
+          false, false, false, false, false);
     } finally {
       zkClient.close();
     }
@@ -183,7 +192,9 @@
     
     SolrZkClient zkClient = new SolrZkClientUsingVMParamsProvidersButWithDifferentVMParamsNames(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     try {
-      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient, true, true, true, true, true);
+      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient,
+          true, true, true, true, true,
+          true, true, true, true, true);
     } finally {
       zkClient.close();
     }
@@ -195,7 +206,9 @@
     
     SolrZkClient zkClient = new SolrZkClientUsingVMParamsProvidersButWithDifferentVMParamsNames(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     try {
-      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient, true, true, false, false, false);
+      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient,
+          true, true, false, false, false,
+          false, false, false, false, false);
     } finally {
       zkClient.close();
     }
@@ -240,28 +253,18 @@
 
         @Override
         public ZkACLProvider createZkACLProvider() {
-          return new DefaultZkACLProvider() {
+          return new VMParamsAllAndReadonlyDigestZkACLProvider() {
             @Override
-            protected List<ACL> createGlobalACLsToAdd() {
-              try {
-                List<ACL> result = new ArrayList<ACL>();
-            
-                if (!StringUtils.isEmpty(digestUsername) && !StringUtils.isEmpty(digestPassword)) {
-                  result.add(new ACL(ZooDefs.Perms.ALL, new Id("digest", DigestAuthenticationProvider.generateDigest(digestUsername + ":" + digestPassword))));
-                }
-            
-                if (!StringUtils.isEmpty(digestReadonlyUsername) && !StringUtils.isEmpty(digestReadonlyPassword)) {
-                  result.add(new ACL(ZooDefs.Perms.READ, new Id("digest", DigestAuthenticationProvider.generateDigest(digestReadonlyUsername + ":" + digestReadonlyPassword))));
-                }
-                
-                if (result.isEmpty()) {
-                  result = ZooDefs.Ids.OPEN_ACL_UNSAFE;
-                }
-                
-                return result;
-              } catch (NoSuchAlgorithmException e) {
-                throw new RuntimeException(e);
-              }
+            protected List<ACL> createNonSecurityACLsToAdd() {
+              return createACLsToAdd(true, digestUsername, digestPassword, digestReadonlyUsername, digestReadonlyPassword);
+            }
+
+            /**
+             * @return Set of ACLs to return security-related znodes
+             */
+            @Override
+            protected List<ACL> createSecurityACLsToAdd() {
+              return createACLsToAdd(false, digestUsername, digestPassword, digestReadonlyUsername, digestReadonlyPassword);
             }
           };
         }
diff --git a/solr/core/src/test/org/apache/solr/cloud/SaslZkACLProviderTest.java b/solr/core/src/test/org/apache/solr/cloud/SaslZkACLProviderTest.java
index 9381c03..16b67a3 100644
--- a/solr/core/src/test/org/apache/solr/cloud/SaslZkACLProviderTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/SaslZkACLProviderTest.java
@@ -20,15 +20,12 @@
 import java.io.IOException;
 import java.lang.invoke.MethodHandles;
 import java.nio.charset.Charset;
-import java.util.Locale;
 
-import javax.security.auth.login.Configuration;
-
-import org.apache.hadoop.minikdc.MiniKdc;
 import org.apache.lucene.util.Constants;
 import org.apache.solr.SolrTestCaseJ4;
 import org.apache.solr.common.cloud.DefaultZkACLProvider;
 import org.apache.solr.common.cloud.SaslZkACLProvider;
+import org.apache.solr.common.cloud.SecurityAwareZkACLProvider;
 import org.apache.solr.common.cloud.SolrZkClient;
 import org.apache.solr.common.cloud.ZkACLProvider;
 import org.apache.solr.util.BadZookeeperThreadsFilter;
@@ -50,8 +47,6 @@
 
   private static final Charset DATA_ENCODING = Charset.forName("UTF-8");
 
-  protected Locale savedLocale = null;
-
   protected ZkTestServer zkServer;
 
   @BeforeClass
@@ -71,7 +66,6 @@
   @Override
   public void setUp() throws Exception {
     super.setUp();
-    savedLocale = KerberosTestUtil.overrideLocaleIfNotSpportedByMiniKdc();
     log.info("####SETUP_START " + getTestName());
     createTempDir();
 
@@ -99,6 +93,7 @@
     try {
       zkClient.create("/protectedCreateNode", "content".getBytes(DATA_ENCODING), CreateMode.PERSISTENT, false);
       zkClient.makePath("/protectedMakePathNode", "content".getBytes(DATA_ENCODING), CreateMode.PERSISTENT, false);
+      zkClient.create(SecurityAwareZkACLProvider.SECURITY_ZNODE_PATH, "content".getBytes(DATA_ENCODING), CreateMode.PERSISTENT, false);
     } finally {
       zkClient.close();
     }
@@ -115,7 +110,6 @@
   @Override
   public void tearDown() throws Exception {
     zkServer.shutdown();
-    Locale.setDefault(savedLocale);
     super.tearDown();
   }
 
@@ -124,7 +118,9 @@
     // Test with Sasl enabled
     SolrZkClient zkClient = new SolrZkClientWithACLs(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     try {
-      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient, true, true, true, true, true);
+      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient,
+          true, true, true, true, true,
+          true, true, true, true, true);
      } finally {
       zkClient.close();
     }
@@ -134,7 +130,9 @@
     System.setProperty("zookeeper.sasl.client", "false");
     zkClient = new SolrZkClientNoACLs(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     try {
-      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient, true, true, false, false, false);
+      VMParamsZkACLAndCredentialsProvidersTest.doTest(zkClient,
+          true, true, false, false, false,
+          false, false, false, false, false);
     } finally {
       zkClient.close();
       System.clearProperty("zookeeper.sasl.client");
@@ -176,8 +174,7 @@
    */
   public static class SaslZkTestServer extends ZkTestServer {
     private String kdcDir;
-    private MiniKdc kdc;
-    private Configuration conf;
+    private KerberosTestServices kerberosTestServices;
 
     public SaslZkTestServer(String zkDir, String kdcDir) {
       super(zkDir);
@@ -187,13 +184,11 @@
     public SaslZkTestServer(String zkDir, int port, String kdcDir) {
       super(zkDir, port);
       this.kdcDir = kdcDir;
-      conf = Configuration.getConfiguration();
     }
 
     @Override
     public void run() throws InterruptedException {
       try {
-        kdc = KerberosTestUtil.getKdc(new File(kdcDir));
         // Don't require that credentials match the entire principal string, e.g.
         // can match "solr" rather than "solr/host@DOMAIN"
         System.setProperty("zookeeper.kerberos.removeRealmFromPrincipal", "true");
@@ -202,12 +197,13 @@
         String zkClientPrincipal = "solr";
         String zkServerPrincipal = "zookeeper/127.0.0.1";
 
-        kdc.start();
-        // Create ZK client and server principals and load them into the Configuration
-        kdc.createPrincipal(keytabFile, zkClientPrincipal, zkServerPrincipal);
-        KerberosTestUtil.JaasConfiguration jaas = new KerberosTestUtil.JaasConfiguration(
-        zkClientPrincipal, keytabFile, zkServerPrincipal, keytabFile);
-        Configuration.setConfiguration(jaas);
+        kerberosTestServices = KerberosTestServices.builder()
+            .withKdc(new File(kdcDir))
+            .withJaasConfiguration(zkClientPrincipal, keytabFile, zkServerPrincipal, keytabFile)
+            .build();
+        kerberosTestServices.start();
+
+        kerberosTestServices.getKdc().createPrincipal(keytabFile, zkClientPrincipal, zkServerPrincipal);
       } catch (Exception ex) {
         throw new RuntimeException(ex);
       }
@@ -220,8 +216,7 @@
       System.clearProperty("zookeeper.authProvider.1");
       System.clearProperty("zookeeper.kerberos.removeRealmFromPrincipal");
       System.clearProperty("zookeeper.kerberos.removeHostFromPrincipal");
-      Configuration.setConfiguration(conf);
-      kdc.stop();
+      kerberosTestServices.stop();
     }
   }
 }
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestAuthenticationFramework.java b/solr/core/src/test/org/apache/solr/cloud/TestAuthenticationFramework.java
index 08db692..6b562a1 100644
--- a/solr/core/src/test/org/apache/solr/cloud/TestAuthenticationFramework.java
+++ b/solr/core/src/test/org/apache/solr/cloud/TestAuthenticationFramework.java
@@ -235,21 +235,23 @@
     public void init(Map<String,Object> pluginConfig) {}
 
     @Override
-    public void doAuthenticate(ServletRequest request, ServletResponse response, FilterChain filterChain)
+    public boolean doAuthenticate(ServletRequest request, ServletResponse response, FilterChain filterChain)
         throws Exception {
       if (expectedUsername == null) {
         filterChain.doFilter(request, response);
-        return;
+        return true;
       }
       HttpServletRequest httpRequest = (HttpServletRequest)request;
       String username = httpRequest.getHeader("username");
       String password = httpRequest.getHeader("password");
       
       log.info("Username: "+username+", password: "+password);
-      if(MockAuthenticationPlugin.expectedUsername.equals(username) && MockAuthenticationPlugin.expectedPassword.equals(password))      
+      if(MockAuthenticationPlugin.expectedUsername.equals(username) && MockAuthenticationPlugin.expectedPassword.equals(password)) {
         filterChain.doFilter(request, response);
-      else {
+        return true;
+      } else {
         ((HttpServletResponse)response).sendError(401, "Unauthorized request");
+        return false;
       }
     }
 
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestMiniSolrCloudClusterKerberos.java b/solr/core/src/test/org/apache/solr/cloud/TestMiniSolrCloudClusterKerberos.java
index 37439b0..d644967 100644
--- a/solr/core/src/test/org/apache/solr/cloud/TestMiniSolrCloudClusterKerberos.java
+++ b/solr/core/src/test/org/apache/solr/cloud/TestMiniSolrCloudClusterKerberos.java
@@ -16,16 +16,13 @@
  */
 package org.apache.solr.cloud;
 
-import javax.security.auth.login.Configuration;
 import java.io.File;
 import java.nio.charset.StandardCharsets;
-import java.util.Locale;
 
 import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
 import com.carrotsearch.randomizedtesting.rules.SystemPropertiesRestoreRule;
 
 import org.apache.commons.io.FileUtils;
-import org.apache.hadoop.minikdc.MiniKdc;
 import org.apache.lucene.util.LuceneTestCase;
 import org.apache.lucene.util.LuceneTestCase.SuppressSysoutChecks;
 import org.apache.solr.util.BadZookeeperThreadsFilter;
@@ -52,17 +49,14 @@
 @SuppressSysoutChecks(bugUrl = "Solr logs to JUL")
 public class TestMiniSolrCloudClusterKerberos extends TestMiniSolrCloudCluster {
 
-  private final Configuration originalConfig = Configuration.getConfiguration();
-
   public TestMiniSolrCloudClusterKerberos () {
     NUM_SERVERS = 5;
     NUM_SHARDS = 2;
     REPLICATION_FACTOR = 2;
   }
   
-  private MiniKdc kdc;
+  private KerberosTestServices kerberosTestServices;
 
-  private Locale savedLocale; // in case locale is broken and we need to fill in a working locale
   @Rule
   public TestRule solrTestRules = RuleChain
       .outerRule(new SystemPropertiesRestoreRule());
@@ -74,20 +68,22 @@
 
   @Override
   public void setUp() throws Exception {
-    savedLocale = KerberosTestUtil.overrideLocaleIfNotSpportedByMiniKdc();
     super.setUp();
     setupMiniKdc();
   }
   
   private void setupMiniKdc() throws Exception {
     String kdcDir = createTempDir()+File.separator+"minikdc";
-    kdc = KerberosTestUtil.getKdc(new File(kdcDir));
     File keytabFile = new File(kdcDir, "keytabs");
     String principal = "HTTP/127.0.0.1";
     String zkServerPrincipal = "zookeeper/127.0.0.1";
+    KerberosTestServices kerberosTestServices = KerberosTestServices.builder()
+        .withKdc(new File(kdcDir))
+        .withJaasConfiguration(principal, keytabFile, zkServerPrincipal, keytabFile)
+        .build();
 
-    kdc.start();
-    kdc.createPrincipal(keytabFile, principal, zkServerPrincipal);
+    kerberosTestServices.start();
+    kerberosTestServices.getKdc().createPrincipal(keytabFile, principal, zkServerPrincipal);
 
     String jaas = "Client {\n"
         + " com.sun.security.auth.module.Krb5LoginModule required\n"
@@ -109,10 +105,7 @@
         + " debug=true\n"
         + " principal=\""+zkServerPrincipal+"\";\n" 
         + "};\n";
-    
-    Configuration conf = new KerberosTestUtil.JaasConfiguration(principal, keytabFile, zkServerPrincipal, keytabFile);
-    javax.security.auth.login.Configuration.setConfiguration(conf);
-    
+
     String jaasFilePath = kdcDir+File.separator + "jaas-client.conf";
     FileUtils.write(new File(jaasFilePath), jaas, StandardCharsets.UTF_8);
     System.setProperty("java.security.auth.login.config", jaasFilePath);
@@ -156,11 +149,7 @@
     System.clearProperty("kerberos.principal");
     System.clearProperty("kerberos.keytab");
     System.clearProperty("authenticationPlugin");
-    Configuration.setConfiguration(this.originalConfig);
-    if (kdc != null) {
-      kdc.stop();
-    }
-    Locale.setDefault(savedLocale);
+    kerberosTestServices.stop();
     super.tearDown();
   }
 }
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestSolrCloudWithDelegationTokens.java b/solr/core/src/test/org/apache/solr/cloud/TestSolrCloudWithDelegationTokens.java
new file mode 100644
index 0000000..ae1c439
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/TestSolrCloudWithDelegationTokens.java
@@ -0,0 +1,405 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.solr.cloud;
+
+import junit.framework.Assert;
+import org.apache.hadoop.util.Time;
+import org.apache.lucene.util.LuceneTestCase;
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.impl.HttpSolrClient;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.embedded.JettySolrRunner;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.DelegationTokenRequest;
+import org.apache.solr.client.solrj.response.DelegationTokenResponse;
+import org.apache.solr.common.SolrException.ErrorCode;
+import org.apache.solr.common.cloud.SolrZkClient;
+import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import static org.apache.solr.security.HttpParamDelegationTokenAuthenticationHandler.USER_PARAM;
+
+import org.apache.http.HttpStatus;
+import org.apache.solr.security.HttpParamDelegationTokenAuthenticationHandler;
+import org.apache.solr.security.KerberosPlugin;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.lang.invoke.MethodHandles;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test the delegation token support in the {@link org.apache.solr.security.KerberosPlugin}.
+ */
+@LuceneTestCase.Slow
+public class TestSolrCloudWithDelegationTokens extends SolrTestCaseJ4 {
+  private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+  private static final int NUM_SERVERS = 2;
+  private static MiniSolrCloudCluster miniCluster;
+  private static HttpSolrClient solrClientPrimary;
+  private static HttpSolrClient solrClientSecondary;
+
+  @BeforeClass
+  public static void startup() throws Exception {
+    System.setProperty("authenticationPlugin", KerberosPlugin.class.getName());
+    System.setProperty(KerberosPlugin.DELEGATION_TOKEN_ENABLED, "true");
+    System.setProperty(KerberosPlugin.AUTH_HANDLER_PARAM,
+        HttpParamDelegationTokenAuthenticationHandler.class.getName());
+    System.setProperty("solr.kerberos.cookie.domain", "127.0.0.1");
+
+    miniCluster = new MiniSolrCloudCluster(NUM_SERVERS, createTempDir(), buildJettyConfig("/solr"));
+    JettySolrRunner runnerPrimary = miniCluster.getJettySolrRunners().get(0);
+    solrClientPrimary =
+        new HttpSolrClient.Builder(runnerPrimary.getBaseUrl().toString())
+            .build();
+    JettySolrRunner runnerSecondary = miniCluster.getJettySolrRunners().get(1);
+    solrClientSecondary =
+        new HttpSolrClient.Builder(runnerSecondary.getBaseUrl().toString())
+            .build();
+  }
+
+  @AfterClass
+  public static void shutdown() throws Exception {
+    if (miniCluster != null) {
+      miniCluster.shutdown();
+    }
+    miniCluster = null;
+    solrClientPrimary.close();
+    solrClientPrimary = null;
+    solrClientSecondary.close();
+    solrClientSecondary = null;
+    System.clearProperty("authenticationPlugin");
+    System.clearProperty(KerberosPlugin.DELEGATION_TOKEN_ENABLED);
+    System.clearProperty(KerberosPlugin.AUTH_HANDLER_PARAM);
+    System.clearProperty("solr.kerberos.cookie.domain");
+  }
+
+  private String getDelegationToken(final String renewer, final String user, HttpSolrClient solrClient) throws Exception {
+    DelegationTokenRequest.Get get = new DelegationTokenRequest.Get(renewer) {
+      @Override
+      public SolrParams getParams() {
+        ModifiableSolrParams params = new ModifiableSolrParams(super.getParams());
+        params.set(USER_PARAM, user);
+        return params;
+      }
+    };
+    DelegationTokenResponse.Get getResponse = get.process(solrClient);
+    return getResponse.getDelegationToken();
+  }
+
+  private long renewDelegationToken(final String token, final int expectedStatusCode,
+      final String user, HttpSolrClient client) throws Exception {
+    DelegationTokenRequest.Renew renew = new DelegationTokenRequest.Renew(token) {
+      @Override
+      public SolrParams getParams() {
+        ModifiableSolrParams params = new ModifiableSolrParams(super.getParams());
+        params.set(USER_PARAM, user);
+        return params;
+      }
+
+      @Override
+      public Set<String> getQueryParams() {
+        Set<String> queryParams = super.getQueryParams();
+        queryParams.add(USER_PARAM);
+        return queryParams;
+      }
+    };
+    try {
+      DelegationTokenResponse.Renew renewResponse = renew.process(client);
+      assertEquals(HttpStatus.SC_OK, expectedStatusCode);
+      return renewResponse.getExpirationTime();
+    } catch (HttpSolrClient.RemoteSolrException ex) {
+      assertEquals(expectedStatusCode, ex.code());
+      return -1;
+    }
+  }
+
+  private void cancelDelegationToken(String token, int expectedStatusCode, HttpSolrClient client)
+  throws Exception {
+    DelegationTokenRequest.Cancel cancel = new DelegationTokenRequest.Cancel(token);
+    try {
+      cancel.process(client);
+      assertEquals(HttpStatus.SC_OK, expectedStatusCode);
+    } catch (HttpSolrClient.RemoteSolrException ex) {
+      assertEquals(expectedStatusCode, ex.code());
+    }
+  }
+
+  private void doSolrRequest(String token, int expectedStatusCode, HttpSolrClient client)
+  throws Exception {
+    doSolrRequest(token, expectedStatusCode, client, 1);
+  }
+
+  private void doSolrRequest(String token, int expectedStatusCode, HttpSolrClient client, int trials)
+  throws Exception {
+    int lastStatusCode = 0;
+    for (int i = 0; i < trials; ++i) {
+      lastStatusCode = getStatusCode(token, null, null, client);
+      if (lastStatusCode == expectedStatusCode) {
+        return;
+      }
+      Thread.sleep(1000);
+    }
+    assertEquals("Did not receieve excepted status code", expectedStatusCode, lastStatusCode);
+  }
+
+  private SolrRequest getAdminRequest(final SolrParams params) {
+    return new CollectionAdminRequest.List() {
+      @Override
+      public SolrParams getParams() {
+        ModifiableSolrParams p = new ModifiableSolrParams(super.getParams());
+        p.add(params);
+        return p;
+      }
+    };
+  }
+
+  private int getStatusCode(String token, final String user, final String op, HttpSolrClient client)
+  throws Exception {
+    HttpSolrClient delegationTokenServer =
+        new HttpSolrClient.Builder(client.getBaseURL().toString())
+            .withDelegationToken(token)
+            .withResponseParser(client.getParser())
+            .build();
+    try {
+      ModifiableSolrParams p = new ModifiableSolrParams();
+      if (user != null) p.set(USER_PARAM, user);
+      if (op != null) p.set("op", op);
+      SolrRequest req = getAdminRequest(p);
+      if (user != null || op != null) {
+        Set<String> queryParams = new HashSet<String>();
+        if (user != null) queryParams.add(USER_PARAM);
+        if (op != null) queryParams.add("op");
+        req.setQueryParams(queryParams);
+      }
+      try {
+        delegationTokenServer.request(req, null, null);
+        return HttpStatus.SC_OK;
+      } catch (HttpSolrClient.RemoteSolrException re) {
+        return re.code();
+      }
+    } finally {
+      delegationTokenServer.close();
+    }
+  }
+
+  private void doSolrRequest(HttpSolrClient client, SolrRequest request,
+      int expectedStatusCode) throws Exception {
+    try {
+      client.request(request);
+      assertEquals(HttpStatus.SC_OK, expectedStatusCode);
+    } catch (HttpSolrClient.RemoteSolrException ex) {
+      assertEquals(expectedStatusCode, ex.code());
+    }
+  }
+
+  private void verifyTokenValid(String token) throws Exception {
+     // pass with token
+    doSolrRequest(token, HttpStatus.SC_OK, solrClientPrimary);
+
+    // fail without token
+    doSolrRequest(null, ErrorCode.UNAUTHORIZED.code, solrClientPrimary);
+
+    // pass with token on other server
+    doSolrRequest(token, HttpStatus.SC_OK, solrClientSecondary);
+
+    // fail without token on other server
+    doSolrRequest(null, ErrorCode.UNAUTHORIZED.code, solrClientSecondary);
+  }
+
+  /**
+   * Test basic Delegation Token get/verify
+   */
+  @Test
+  public void testDelegationTokenVerify() throws Exception {
+    final String user = "bar";
+
+    // Get token
+    String token = getDelegationToken(null, user, solrClientPrimary);
+    assertNotNull(token);
+    verifyTokenValid(token);
+  }
+
+  private void verifyTokenCancelled(String token, HttpSolrClient client) throws Exception {
+    // fail with token on both servers.  If cancelToOtherURL is true,
+    // the request went to other url, so FORBIDDEN should be returned immediately.
+    // The cancelled token may take awhile to propogate to the standard url (via ZK).
+    // This is of course the opposite if cancelToOtherURL is false.
+    doSolrRequest(token, ErrorCode.FORBIDDEN.code, client, 10);
+
+    // fail without token on both servers
+    doSolrRequest(null, ErrorCode.UNAUTHORIZED.code, solrClientPrimary);
+    doSolrRequest(null, ErrorCode.UNAUTHORIZED.code, solrClientSecondary);
+  }
+
+  @Test
+  public void testDelegationTokenCancel() throws Exception {
+    {
+      // Get token
+      String token = getDelegationToken(null, "user", solrClientPrimary);
+      assertNotNull(token);
+
+      // cancel token, note don't need to be authenticated to cancel (no user specified)
+      cancelDelegationToken(token, HttpStatus.SC_OK, solrClientPrimary);
+      verifyTokenCancelled(token, solrClientPrimary);
+    }
+
+    {
+      // cancel token on different server from where we got it
+      String token = getDelegationToken(null, "user", solrClientPrimary);
+      assertNotNull(token);
+
+      cancelDelegationToken(token, HttpStatus.SC_OK, solrClientSecondary);
+      verifyTokenCancelled(token, solrClientSecondary);
+    }
+  }
+
+  @Test
+  public void testDelegationTokenCancelFail() throws Exception {
+    // cancel a bogus token
+    cancelDelegationToken("BOGUS", ErrorCode.NOT_FOUND.code, solrClientPrimary);
+
+    {
+      // cancel twice, first on same server
+      String token = getDelegationToken(null, "bar", solrClientPrimary);
+      assertNotNull(token);
+      cancelDelegationToken(token, HttpStatus.SC_OK, solrClientPrimary);
+      cancelDelegationToken(token, ErrorCode.NOT_FOUND.code, solrClientSecondary);
+      cancelDelegationToken(token, ErrorCode.NOT_FOUND.code, solrClientPrimary);
+    }
+
+    {
+      // cancel twice, first on other server
+      String token = getDelegationToken(null, "bar", solrClientPrimary);
+      assertNotNull(token);
+      cancelDelegationToken(token, HttpStatus.SC_OK, solrClientSecondary);
+      cancelDelegationToken(token, ErrorCode.NOT_FOUND.code, solrClientSecondary);
+      cancelDelegationToken(token, ErrorCode.NOT_FOUND.code, solrClientPrimary);
+    }
+  }
+
+  private void verifyDelegationTokenRenew(String renewer, String user)
+  throws Exception {
+    {
+      // renew on same server
+      String token = getDelegationToken(renewer, user, solrClientPrimary);
+      assertNotNull(token);
+      long now = Time.now();
+      assertTrue(renewDelegationToken(token, HttpStatus.SC_OK, user, solrClientPrimary) > now);
+      verifyTokenValid(token);
+    }
+
+    {
+      // renew on different server
+      String token = getDelegationToken(renewer, user, solrClientPrimary);
+      assertNotNull(token);
+      long now = Time.now();
+      assertTrue(renewDelegationToken(token, HttpStatus.SC_OK, user, solrClientSecondary) > now);
+      verifyTokenValid(token);
+    }
+  }
+
+  @Test
+  public void testDelegationTokenRenew() throws Exception {
+    // test with specifying renewer
+    verifyDelegationTokenRenew("bar", "bar");
+
+    // test without specifying renewer
+    verifyDelegationTokenRenew(null, "bar");
+  }
+
+  @Test
+  public void testDelegationTokenRenewFail() throws Exception {
+    // don't set renewer and try to renew as an a different user
+    String token = getDelegationToken(null, "bar", solrClientPrimary);
+    assertNotNull(token);
+    renewDelegationToken(token, ErrorCode.FORBIDDEN.code, "foo", solrClientPrimary);
+    renewDelegationToken(token, ErrorCode.FORBIDDEN.code, "foo", solrClientSecondary);
+
+    // set renewer and try to renew as different user
+    token = getDelegationToken("renewUser", "bar", solrClientPrimary);
+    assertNotNull(token);
+    renewDelegationToken(token, ErrorCode.FORBIDDEN.code, "notRenewUser", solrClientPrimary);
+    renewDelegationToken(token, ErrorCode.FORBIDDEN.code, "notRenewUser", solrClientSecondary);
+  }
+
+  /**
+   * Test that a non-delegation-token "op" http param is handled correctly
+   */
+  @Test
+  public void testDelegationOtherOp() throws Exception {
+    assertEquals(HttpStatus.SC_OK, getStatusCode(null, "bar", "someSolrOperation", solrClientPrimary));
+  }
+
+  @Test
+  public void testZNodePaths() throws Exception {
+    getDelegationToken(null, "bar", solrClientPrimary);
+    SolrZkClient zkClient = new SolrZkClient(miniCluster.getZkServer().getZkAddress(), 1000);
+    try {
+      assertTrue(zkClient.exists("/security/zkdtsm", true));
+      assertTrue(zkClient.exists("/security/token", true));
+    } finally {
+      zkClient.close();
+    }
+  }
+
+  /**
+   * Test HttpSolrServer's delegation token support
+   */
+  @Test
+  public void testDelegationTokenSolrClient() throws Exception {
+    // Get token
+    String token = getDelegationToken(null, "bar", solrClientPrimary);
+    assertNotNull(token);
+
+    SolrRequest request = getAdminRequest(new ModifiableSolrParams());
+
+    // test without token
+    HttpSolrClient ss =
+        new HttpSolrClient.Builder(solrClientPrimary.getBaseURL().toString())
+            .withResponseParser(solrClientPrimary.getParser())
+            .build();
+    try {
+      doSolrRequest(ss, request, ErrorCode.UNAUTHORIZED.code);
+    } finally {
+      ss.close();
+    }
+
+    ss = new HttpSolrClient.Builder(solrClientPrimary.getBaseURL().toString())
+        .withDelegationToken(token)
+        .withResponseParser(solrClientPrimary.getParser())
+        .build();
+    try {
+      // test with token via property
+      doSolrRequest(ss, request, HttpStatus.SC_OK);
+
+      // test with param -- should throw an exception
+      ModifiableSolrParams tokenParam = new ModifiableSolrParams();
+      tokenParam.set("delegation", "invalidToken");
+      try {
+        doSolrRequest(ss, getAdminRequest(tokenParam), ErrorCode.FORBIDDEN.code);
+        Assert.fail("Expected exception");
+      } catch (IllegalArgumentException ex) {}
+    } finally {
+      ss.close();
+    }
+  }
+}
diff --git a/solr/core/src/test/org/apache/solr/cloud/TestSolrCloudWithKerberosAlt.java b/solr/core/src/test/org/apache/solr/cloud/TestSolrCloudWithKerberosAlt.java
index 6ac2254..c505b51 100644
--- a/solr/core/src/test/org/apache/solr/cloud/TestSolrCloudWithKerberosAlt.java
+++ b/solr/core/src/test/org/apache/solr/cloud/TestSolrCloudWithKerberosAlt.java
@@ -17,15 +17,12 @@
 package org.apache.solr.cloud;
 
 import java.io.File;
+import java.lang.invoke.MethodHandles;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
-import java.util.Locale;
 import java.util.Properties;
 
-import javax.security.auth.login.Configuration;
-
 import org.apache.commons.io.FileUtils;
-import org.apache.hadoop.minikdc.MiniKdc;
 import org.apache.lucene.index.TieredMergePolicy;
 import org.apache.lucene.util.Constants;
 import org.apache.lucene.util.LuceneTestCase;
@@ -49,6 +46,8 @@
 import org.junit.Test;
 import org.junit.rules.RuleChain;
 import org.junit.rules.TestRule;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
 import com.carrotsearch.randomizedtesting.rules.SystemPropertiesRestoreRule;
@@ -67,7 +66,7 @@
 @LuceneTestCase.SuppressSysoutChecks(bugUrl = "Solr logs to JUL")
 public class TestSolrCloudWithKerberosAlt extends LuceneTestCase {
 
-  private final Configuration originalConfig = Configuration.getConfiguration();
+  private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
   protected final int NUM_SERVERS;
   protected final int NUM_SHARDS;
   protected final int REPLICATION_FACTOR;
@@ -78,10 +77,8 @@
     REPLICATION_FACTOR = 1;
   }
 
-  private MiniKdc kdc;
+  private KerberosTestServices kerberosTestServices;
 
-  private Locale savedLocale; // in case locale is broken and we need to fill in a working locale
-  
   @Rule
   public TestRule solrTestRules = RuleChain
       .outerRule(new SystemPropertiesRestoreRule());
@@ -98,7 +95,6 @@
 
   @Override
   public void setUp() throws Exception {
-    savedLocale = KerberosTestUtil.overrideLocaleIfNotSpportedByMiniKdc();
     super.setUp();
     setupMiniKdc();
   }
@@ -106,12 +102,15 @@
   private void setupMiniKdc() throws Exception {
     System.setProperty("solr.jaas.debug", "true");
     String kdcDir = createTempDir()+File.separator+"minikdc";
-    kdc = KerberosTestUtil.getKdc(new File(kdcDir));
-    File keytabFile = new File(kdcDir, "keytabs");
-    String solrServerPrincipal = "HTTP/127.0.0.1";
     String solrClientPrincipal = "solr";
-    kdc.start();
-    kdc.createPrincipal(keytabFile, solrServerPrincipal, solrClientPrincipal);
+    File keytabFile = new File(kdcDir, "keytabs");
+    kerberosTestServices = KerberosTestServices.builder()
+        .withKdc(new File(kdcDir))
+        .withJaasConfiguration(solrClientPrincipal, keytabFile, "SolrClient")
+        .build();
+    String solrServerPrincipal = "HTTP/127.0.0.1";
+    kerberosTestServices.start();
+    kerberosTestServices.getKdc().createPrincipal(keytabFile, solrServerPrincipal, solrClientPrincipal);
 
     String jaas = "SolrClient {\n"
         + " com.sun.security.auth.module.Krb5LoginModule required\n"
@@ -124,9 +123,6 @@
         + " principal=\"" + solrClientPrincipal + "\";\n"
         + "};";
 
-    Configuration conf = new KerberosTestUtil.JaasConfiguration(solrClientPrincipal, keytabFile, "SolrClient");
-    Configuration.setConfiguration(conf);
-
     String jaasFilePath = kdcDir+File.separator+"jaas-client.conf";
     FileUtils.write(new File(jaasFilePath), jaas, StandardCharsets.UTF_8);
     System.setProperty("java.security.auth.login.config", jaasFilePath);
@@ -135,6 +131,9 @@
     System.setProperty("solr.kerberos.principal", solrServerPrincipal);
     System.setProperty("solr.kerberos.keytab", keytabFile.getAbsolutePath());
     System.setProperty("authenticationPlugin", "org.apache.solr.security.KerberosPlugin");
+    boolean enableDt = random().nextBoolean();
+    log.info("Enable delegation token: " + enableDt);
+    System.setProperty("solr.kerberos.delegation.token.enabled", new Boolean(enableDt).toString());
     // Extracts 127.0.0.1 from HTTP/127.0.0.1@EXAMPLE.COM
     System.setProperty("solr.kerberos.name.rules", "RULE:[1:$1@$0](.*EXAMPLE.COM)s/@.*//"
         + "\nRULE:[2:$2@$0](.*EXAMPLE.COM)s/@.*//"
@@ -240,11 +239,7 @@
     System.clearProperty("authenticationPlugin");
     System.clearProperty("solr.kerberos.name.rules");
     System.clearProperty("solr.jaas.debug");
-    Configuration.setConfiguration(this.originalConfig);
-    if (kdc != null) {
-      kdc.stop();
-    }
-    Locale.setDefault(savedLocale);
+    kerberosTestServices.stop();
     super.tearDown();
   }
 }
diff --git a/solr/core/src/test/org/apache/solr/cloud/VMParamsZkACLAndCredentialsProvidersTest.java b/solr/core/src/test/org/apache/solr/cloud/VMParamsZkACLAndCredentialsProvidersTest.java
index 31919a8..95422fa 100644
--- a/solr/core/src/test/org/apache/solr/cloud/VMParamsZkACLAndCredentialsProvidersTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/VMParamsZkACLAndCredentialsProvidersTest.java
@@ -21,6 +21,7 @@
 import java.nio.charset.Charset;
 
 import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.common.cloud.SecurityAwareZkACLProvider;
 import org.apache.solr.common.cloud.SolrZkClient;
 import org.apache.solr.common.cloud.VMParamsAllAndReadonlyDigestZkACLProvider;
 import org.apache.solr.common.cloud.VMParamsSingleSetCredentialsDigestZkCredentialsProvider;
@@ -76,6 +77,8 @@
     zkClient = new SolrZkClient(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     zkClient.create("/protectedCreateNode", "content".getBytes(DATA_ENCODING), CreateMode.PERSISTENT, false);
     zkClient.makePath("/protectedMakePathNode", "content".getBytes(DATA_ENCODING), CreateMode.PERSISTENT, false);
+
+    zkClient.create(SecurityAwareZkACLProvider.SECURITY_ZNODE_PATH, "content".getBytes(DATA_ENCODING), CreateMode.PERSISTENT, false);
     zkClient.close();
     
     clearSecuritySystemProperties();
@@ -106,7 +109,9 @@
     
     SolrZkClient zkClient = new SolrZkClient(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     try {
-      doTest(zkClient, false, false, false, false, false);
+      doTest(zkClient,
+          false, false, false, false, false,
+          false, false, false, false, false);
     } finally {
       zkClient.close();
     }
@@ -118,7 +123,9 @@
     
     SolrZkClient zkClient = new SolrZkClient(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     try {
-      doTest(zkClient, false, false, false, false, false);
+      doTest(zkClient,
+          false, false, false, false, false,
+          false, false, false, false, false);
     } finally {
       zkClient.close();
     }
@@ -130,7 +137,9 @@
 
     SolrZkClient zkClient = new SolrZkClient(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     try {
-      doTest(zkClient, true, true, true, true, true);
+      doTest(zkClient,
+          true, true, true, true, true,
+          true, true, true, true, true);
     } finally {
       zkClient.close();
     }
@@ -142,17 +151,23 @@
 
     SolrZkClient zkClient = new SolrZkClient(zkServer.getZkAddress(), AbstractZkTestCase.TIMEOUT);
     try {
-      doTest(zkClient, true, true, false, false, false);
+      doTest(zkClient,
+          true, true, false, false, false,
+          false, false, false, false, false);
     } finally {
       zkClient.close();
     }
   }
     
-  protected static void doTest(SolrZkClient zkClient, boolean getData, boolean list, boolean create, boolean setData, boolean delete) throws Exception {
+  protected static void doTest(
+      SolrZkClient zkClient,
+      boolean getData, boolean list, boolean create, boolean setData, boolean delete,
+      boolean secureGet, boolean secureList, boolean secureCreate, boolean secureSet, boolean secureDelete) throws Exception {
     doTest(zkClient, "/protectedCreateNode", getData, list, create, setData, delete);
     doTest(zkClient, "/protectedMakePathNode", getData, list, create, setData, delete);
     doTest(zkClient, "/unprotectedCreateNode", true, true, true, true, delete);
     doTest(zkClient, "/unprotectedMakePathNode", true, true, true, true, delete);
+    doTest(zkClient, SecurityAwareZkACLProvider.SECURITY_ZNODE_PATH, secureGet, secureList, secureCreate, secureSet, secureDelete);
   }
   
   protected static void doTest(SolrZkClient zkClient, String path, boolean getData, boolean list, boolean create, boolean setData, boolean delete) throws Exception {
diff --git a/solr/core/src/test/org/apache/solr/security/HttpParamDelegationTokenAuthenticationHandler.java b/solr/core/src/test/org/apache/solr/security/HttpParamDelegationTokenAuthenticationHandler.java
new file mode 100644
index 0000000..7c5c94a
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/security/HttpParamDelegationTokenAuthenticationHandler.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.solr.security;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.hadoop.security.authentication.client.AuthenticationException;
+import org.apache.hadoop.security.authentication.server.AuthenticationHandler;
+import org.apache.hadoop.security.authentication.server.AuthenticationToken;
+import org.apache.hadoop.security.token.delegation.web.DelegationTokenAuthenticationHandler;
+
+import org.apache.http.NameValuePair;
+import org.apache.http.client.utils.URLEncodedUtils;
+
+/**
+ * AuthenticationHandler that supports delegation tokens and simple
+ * authentication via the "user" http parameter
+ */
+public class HttpParamDelegationTokenAuthenticationHandler extends
+    DelegationTokenAuthenticationHandler {
+
+  public static final String USER_PARAM = "user";
+
+  public HttpParamDelegationTokenAuthenticationHandler() {
+    super(new HttpParamAuthenticationHandler());
+  }
+
+  @Override
+  public void init(Properties config) throws ServletException {
+    Properties conf = new Properties();
+    for (Map.Entry entry : config.entrySet()) {
+      conf.setProperty((String) entry.getKey(), (String) entry.getValue());
+    }
+    conf.setProperty(TOKEN_KIND, KerberosPlugin.DELEGATION_TOKEN_TYPE_DEFAULT);
+    super.init(conf);
+  }
+ 
+  private static String getHttpParam(HttpServletRequest request, String param) {
+    List<NameValuePair> pairs =
+      URLEncodedUtils.parse(request.getQueryString(), Charset.forName("UTF-8"));
+    for (NameValuePair nvp : pairs) {
+      if(param.equals(nvp.getName())) {
+        return nvp.getValue();
+      }
+    }
+    return null;
+  }
+
+  private static class HttpParamAuthenticationHandler
+      implements AuthenticationHandler {
+
+    @Override
+    public String getType() {
+      return "dummy";
+    }
+
+    @Override
+    public void init(Properties config) throws ServletException {
+    }
+
+    @Override
+    public void destroy() {
+    }
+
+    @Override
+    public boolean managementOperation(AuthenticationToken token,
+        HttpServletRequest request, HttpServletResponse response)
+        throws IOException, AuthenticationException {
+      return false;
+    }
+
+    @Override
+    public AuthenticationToken authenticate(HttpServletRequest request,
+        HttpServletResponse response)
+        throws IOException, AuthenticationException {
+      AuthenticationToken token = null;
+      String userName = getHttpParam(request, USER_PARAM);
+      if (userName != null) {
+        return new AuthenticationToken(userName, userName, "test");
+      } else {
+        response.setStatus(HttpServletResponse.SC_UNAUTHORIZED);
+        response.setHeader("WWW-Authenticate", "dummy");
+      }
+      return token;
+    }
+  }
+}
diff --git a/solr/core/src/test/org/apache/solr/security/MockAuthenticationPlugin.java b/solr/core/src/test/org/apache/solr/security/MockAuthenticationPlugin.java
index e3cf7bd..3013086 100644
--- a/solr/core/src/test/org/apache/solr/security/MockAuthenticationPlugin.java
+++ b/solr/core/src/test/org/apache/solr/security/MockAuthenticationPlugin.java
@@ -20,11 +20,16 @@
 import javax.servlet.ServletException;
 import javax.servlet.ServletRequest;
 import javax.servlet.ServletResponse;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletRequestWrapper;
 import java.io.IOException;
 import java.security.Principal;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Predicate;
 
+import org.apache.http.auth.BasicUserPrincipal;
+
 public class MockAuthenticationPlugin extends AuthenticationPlugin {
   static Predicate<ServletRequest> predicate;
 
@@ -33,7 +38,7 @@
   }
 
   @Override
-  public void doAuthenticate(ServletRequest request, ServletResponse response, FilterChain filterChain) throws IOException, ServletException {
+  public boolean doAuthenticate(ServletRequest request, ServletResponse response, FilterChain filterChain) throws IOException, ServletException {
     String user = null;
     if (predicate != null) {
       if (predicate.test(request)) {
@@ -41,9 +46,32 @@
         request.removeAttribute(Principal.class.getName());
       }
     }
-    forward(user, request, response, filterChain);
+
+    final FilterChain ffc = filterChain;
+    final AtomicBoolean requestContinues = new AtomicBoolean(false);
+    forward(user, request, response, new FilterChain() {
+      @Override
+      public void doFilter(ServletRequest req, ServletResponse res) throws IOException, ServletException {
+        ffc.doFilter(req, res);
+        requestContinues.set(true);
+      }
+    });
+    return requestContinues.get();
   }
 
+  protected void forward(String user, ServletRequest  req, ServletResponse rsp,
+                                    FilterChain chain) throws IOException, ServletException {
+    if(user != null) {
+      final Principal p = new BasicUserPrincipal(user);
+      req = new HttpServletRequestWrapper((HttpServletRequest) req) {
+        @Override
+        public Principal getUserPrincipal() {
+          return p;
+        }
+      };
+    }
+    chain.doFilter(req,rsp);
+  }
 
   @Override
   public void close() throws IOException {
diff --git a/solr/licenses/curator-recipes-2.8.0.jar.sha1 b/solr/licenses/curator-recipes-2.8.0.jar.sha1
new file mode 100644
index 0000000..82d8946
--- /dev/null
+++ b/solr/licenses/curator-recipes-2.8.0.jar.sha1
@@ -0,0 +1 @@
+c563e25fb37f85a6b029bc9746e75573640474fb
diff --git a/solr/licenses/curator-recipes-LICENSE-ASL.txt b/solr/licenses/curator-recipes-LICENSE-ASL.txt
new file mode 100644
index 0000000..7a4a3ea
--- /dev/null
+++ b/solr/licenses/curator-recipes-LICENSE-ASL.txt
@@ -0,0 +1,202 @@
+
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright [yyyy] [name of copyright owner]
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
\ No newline at end of file
diff --git a/solr/licenses/curator-recipes-NOTICE.txt b/solr/licenses/curator-recipes-NOTICE.txt
new file mode 100644
index 0000000..f568d0f
--- /dev/null
+++ b/solr/licenses/curator-recipes-NOTICE.txt
@@ -0,0 +1,5 @@
+Apache Curator
+Copyright 2013-2014 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
\ No newline at end of file
diff --git a/solr/solrj/ivy.xml b/solr/solrj/ivy.xml
index ceefcc3..f2296c5 100644
--- a/solr/solrj/ivy.xml
+++ b/solr/solrj/ivy.xml
@@ -1,3 +1,4 @@
+<?xml version="1.0" encoding="utf-8"?>
 <!--
    Licensed to the Apache Software Foundation (ASF) under one
    or more contributor license agreements.  See the NOTICE file
@@ -40,6 +41,10 @@
 
     <dependency org="org.slf4j" name="slf4j-log4j12" rev="${/org.slf4j/slf4j-log4j12}" conf="test"/>
 
+    <dependency org="com.fasterxml.jackson.core" name="jackson-annotations"  rev="${/com.fasterxml.jackson.core/jackson-annotations}"   conf="compile"/>
+    <dependency org="com.fasterxml.jackson.core" name="jackson-core" rev="${/com.fasterxml.jackson.core/jackson-core}" conf="compile"/>
+    <dependency org="com.fasterxml.jackson.core" name="jackson-databind" rev="${/com.fasterxml.jackson.core/jackson-databind}" conf="compile"/>
+    <dependency org="com.google.guava" name="guava" rev="${/com.google.guava/guava}" conf="compile"/>
     <exclude org="*" ext="*" matcher="regexp" type="${ivy.exclude.types}"/>
   </dependencies>
 </ivy-module>
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpSolrClient.java
index 9b0cf8d..222119c 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpSolrClient.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpSolrClient.java
@@ -23,6 +23,7 @@
 import java.net.ConnectException;
 import java.net.SocketTimeoutException;
 import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Iterator;
@@ -30,6 +31,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Set;
+import java.util.TreeSet;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 
@@ -743,7 +745,44 @@
       super(code, "Error from server at " + remoteHost + ": " + msg, th);
     }
   }
-  
+
+  private static class DelegationTokenHttpSolrClient extends HttpSolrClient {
+    private final String DELEGATION_TOKEN_PARAM = "delegation";
+    private final String delegationToken;
+
+    public DelegationTokenHttpSolrClient(String baseURL,
+                                         HttpClient client,
+                                         ResponseParser parser,
+                                         boolean allowCompression,
+                                         String delegationToken) {
+      super(baseURL, client, parser, allowCompression);
+      if (delegationToken == null) {
+        throw new IllegalArgumentException("Delegation token cannot be null");
+      }
+      this.delegationToken = delegationToken;
+      setQueryParams(new TreeSet<String>(Arrays.asList(DELEGATION_TOKEN_PARAM)));
+      invariantParams = new ModifiableSolrParams();
+      invariantParams.set(DELEGATION_TOKEN_PARAM, delegationToken);
+    }
+
+    @Override
+    protected HttpRequestBase createMethod(final SolrRequest request, String collection) throws IOException, SolrServerException {
+      SolrParams params = request.getParams();
+      if (params.getParams(DELEGATION_TOKEN_PARAM) != null) {
+        throw new IllegalArgumentException(DELEGATION_TOKEN_PARAM + " parameter not supported");
+      }
+      return super.createMethod(request, collection);
+    }
+
+    @Override
+    public void setQueryParams(Set<String> queryParams) {
+      if (queryParams == null || !queryParams.contains(DELEGATION_TOKEN_PARAM)) {
+        throw new IllegalArgumentException("Query params must contain " + DELEGATION_TOKEN_PARAM);
+      }
+      super.setQueryParams(queryParams);
+    }
+  }
+
   /**
    * Constructs {@link HttpSolrClient} instances from provided configuration.
    */
@@ -752,6 +791,7 @@
     private HttpClient httpClient;
     private ResponseParser responseParser;
     private boolean compression;
+    private String delegationToken;
     
     /**
      * Create a Builder object, based on the provided Solr URL.
@@ -788,7 +828,14 @@
       this.compression = compression;
       return this;
     }
-    
+
+    /**
+     * Use a delegation token for authenticating via the KerberosPlugin
+     */
+    public Builder withDelegationToken(String delegationToken) {
+      this.delegationToken = delegationToken;
+      return this;
+    }
     /**
      * Create a {@link HttpSolrClient} based on provided configuration.
      */
@@ -796,7 +843,11 @@
       if (baseSolrUrl == null) {
         throw new IllegalArgumentException("Cannot create HttpSolrClient without a valid baseSolrUrl!");
       }
-      return new HttpSolrClient(baseSolrUrl, httpClient, responseParser, compression);
+      if (delegationToken == null) {
+        return new HttpSolrClient(baseSolrUrl, httpClient, responseParser, compression);
+      } else {
+        return new DelegationTokenHttpSolrClient(baseSolrUrl, httpClient, responseParser, compression, delegationToken);
+      }
     }
   }
 }
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/Krb5HttpClientBuilder.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/Krb5HttpClientBuilder.java
index 9d5a926..84fe5f9 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/Krb5HttpClientBuilder.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/Krb5HttpClientBuilder.java
@@ -26,6 +26,7 @@
 import javax.security.auth.login.AppConfigurationEntry;
 import javax.security.auth.login.Configuration;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.http.HttpEntity;
 import org.apache.http.HttpEntityEnclosingRequest;
 import org.apache.http.HttpRequestInterceptor;
@@ -51,12 +52,21 @@
   public static final String LOGIN_CONFIG_PROP = "java.security.auth.login.config";
   private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
   
-  private static final Configuration jaasConfig = new SolrJaasConfiguration();
+  private static Configuration jaasConfig = new SolrJaasConfiguration();
 
   public Krb5HttpClientBuilder() {
 
   }
-  
+
+  /**
+   * The jaasConfig is static, which makes it problematic for testing in the same jvm.
+   * Call this function to regenerate the static config (this is not thread safe).
+   */
+  @VisibleForTesting
+  public static void regenerateJaasConfiguration() {
+    jaasConfig = new SolrJaasConfiguration();
+  }
+
   public SolrHttpClientBuilder getBuilder() {
     return getBuilder(HttpClientUtil.getHttpClientBuilder());
   }
@@ -104,9 +114,9 @@
             return null;
           }
         };
-        
+
         HttpClientUtil.setCookiePolicy(SolrPortAwareCookieSpecFactory.POLICY_NAME);
-        
+
         builder.setCookieSpecRegistryProvider(() -> {
           SolrPortAwareCookieSpecFactory cookieFactory = new SolrPortAwareCookieSpecFactory();
 
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/request/DelegationTokenRequest.java b/solr/solrj/src/java/org/apache/solr/client/solrj/request/DelegationTokenRequest.java
new file mode 100644
index 0000000..6d2f0cb
--- /dev/null
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/request/DelegationTokenRequest.java
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.request;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.impl.NoOpResponseParser;
+import org.apache.solr.client.solrj.response.DelegationTokenResponse;
+
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.util.ContentStream;
+
+/**
+ * Class for making Solr delegation token requests.
+ *
+ * @since Solr 6.2
+ */
+public abstract class DelegationTokenRequest
+    <Q extends DelegationTokenRequest<Q,R>, R extends DelegationTokenResponse>
+    extends SolrRequest<R> {
+
+  protected static final String OP_KEY = "op";
+  protected static final String TOKEN_KEY = "token";
+
+  public DelegationTokenRequest(METHOD m) {
+    // path doesn't really matter -- the filter will respond to any path.
+    // setting the path to admin/collections lets us pass through CloudSolrServer
+    // without having to specify a collection (that may not even exist yet).
+    super(m, "/admin/collections");
+  }
+
+  protected abstract Q getThis();
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public Collection<ContentStream> getContentStreams() throws IOException {
+    return null;
+  }
+
+  @Override
+  protected abstract R createResponse(SolrClient client);
+
+  public static class Get extends DelegationTokenRequest<Get, DelegationTokenResponse.Get> {
+    protected String renewer;
+
+    public Get() {
+      this(null);
+    }
+
+    public Get(String renewer) {
+      super(METHOD.GET);
+      this.renewer = renewer;
+      setResponseParser(new DelegationTokenResponse.JsonMapResponseParser());
+      setQueryParams(new TreeSet<String>(Arrays.asList(OP_KEY)));
+    }
+
+    @Override
+    protected Get getThis() {
+      return this;
+    }
+
+    @Override
+    public SolrParams getParams() {
+      ModifiableSolrParams params = new ModifiableSolrParams();
+      params.set(OP_KEY, "GETDELEGATIONTOKEN");
+      if (renewer != null) params.set("renewer", renewer);
+      return params;
+    }
+
+    @Override
+    public DelegationTokenResponse.Get createResponse(SolrClient client) { return new DelegationTokenResponse.Get(); }
+  }
+
+  public static class Renew extends DelegationTokenRequest<Renew, DelegationTokenResponse.Renew> {
+    protected String token;
+
+    @Override
+    protected Renew getThis() {
+      return this;
+    }
+
+    public Renew(String token) {
+      super(METHOD.PUT);
+      this.token = token;
+      setResponseParser(new DelegationTokenResponse.JsonMapResponseParser());
+      setQueryParams(new TreeSet<String>(Arrays.asList(OP_KEY, TOKEN_KEY)));
+    }
+
+    @Override
+    public SolrParams getParams() {
+      ModifiableSolrParams params = new ModifiableSolrParams();
+      params.set(OP_KEY, "RENEWDELEGATIONTOKEN");
+      params.set(TOKEN_KEY, token);
+      return params;
+    }
+
+    @Override
+    public DelegationTokenResponse.Renew createResponse(SolrClient client) { return new DelegationTokenResponse.Renew(); }
+  }
+
+  public static class Cancel extends DelegationTokenRequest<Cancel, DelegationTokenResponse.Cancel> {
+    protected String token;
+
+    public Cancel(String token) {
+      super(METHOD.PUT);
+      this.token = token;
+      setResponseParser(new NoOpResponseParser());
+      Set<String> queryParams = new TreeSet<String>();
+      setQueryParams(new TreeSet<String>(Arrays.asList(OP_KEY, TOKEN_KEY)));
+    }
+
+    @Override
+    protected Cancel getThis() {
+      return this;
+    }
+
+    @Override
+    public SolrParams getParams() {
+      ModifiableSolrParams params = new ModifiableSolrParams();
+      params.set(OP_KEY, "CANCELDELEGATIONTOKEN");
+      params.set(TOKEN_KEY, token);
+      return params;
+    }
+
+    @Override
+    public DelegationTokenResponse.Cancel createResponse(SolrClient client) { return new DelegationTokenResponse.Cancel(); }
+  }
+}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/response/DelegationTokenResponse.java b/solr/solrj/src/java/org/apache/solr/client/solrj/response/DelegationTokenResponse.java
new file mode 100644
index 0000000..c80e2eb
--- /dev/null
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/response/DelegationTokenResponse.java
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.response;
+
+import org.apache.solr.client.solrj.ResponseParser;
+import org.apache.solr.common.SolrException;
+import org.apache.solr.common.util.NamedList;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Reader;
+import java.util.Map;
+
+/**
+ * Delegation Token responses
+ */
+public abstract class DelegationTokenResponse extends SolrResponseBase {
+
+  public static class Get extends DelegationTokenResponse {
+
+    /**
+     * Get the urlString to be used as the delegation token
+     */
+    public String getDelegationToken() {
+      try {
+        Map map = (Map)getResponse().get("Token");
+        if (map != null) {
+          return (String)map.get("urlString");
+        }
+      } catch (ClassCastException e) {
+        throw new SolrException (SolrException.ErrorCode.SERVER_ERROR,
+          "parsing error", e);
+      }
+      return null;
+    }
+  }
+
+  public static class Renew extends DelegationTokenResponse {
+    public Long getExpirationTime() {
+      try {
+        return (Long)getResponse().get("long");
+      } catch (ClassCastException e) {
+        throw new SolrException (SolrException.ErrorCode.SERVER_ERROR,
+          "parsing error", e);
+      }
+    }
+  }
+
+  public static class Cancel extends DelegationTokenResponse {
+  }
+
+  /**
+   * ResponseParser for JsonMaps.  Used for Get and Renew DelegationToken responses.
+   */
+  public static class JsonMapResponseParser extends ResponseParser {
+    @Override
+    public String getWriterType() {
+      return "json";
+    }
+
+    @Override
+    public NamedList<Object> processResponse(InputStream body, String encoding) {
+      ObjectMapper mapper = new ObjectMapper();
+      Map map = null;
+      try {
+        map = mapper.readValue(body, Map.class);
+      } catch (IOException e) {
+        throw new SolrException (SolrException.ErrorCode.SERVER_ERROR,
+          "parsing error", e);
+      }
+      NamedList<Object> list = new NamedList<Object>();
+      list.addAll(map);
+      return list;
+    }
+
+    @Override
+    public NamedList<Object> processResponse(Reader reader) {
+      throw new RuntimeException("Cannot handle character stream");
+    }
+
+    @Override
+    public String getContentType() {
+      return "application/json";
+    }
+
+    @Override
+    public String getVersion() {
+      return "1";
+    }
+  }
+}
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/SaslZkACLProvider.java b/solr/solrj/src/java/org/apache/solr/common/cloud/SaslZkACLProvider.java
index eaccb8c..c67ad12 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/SaslZkACLProvider.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/SaslZkACLProvider.java
@@ -30,19 +30,22 @@
  * configurations have already been set up and will not be modified, or
  * where configuration changes are controlled via Solr APIs.
  */
-public class SaslZkACLProvider extends DefaultZkACLProvider {
+public class SaslZkACLProvider extends SecurityAwareZkACLProvider {
 
   private static String superUser = System.getProperty("solr.authorization.superuser", "solr");
 
   @Override
-  protected List<ACL> createGlobalACLsToAdd() {
-    List<ACL> result = new ArrayList<ACL>();
-    result.add(new ACL(ZooDefs.Perms.ALL, new Id("sasl", superUser)));
-    result.add(new ACL(ZooDefs.Perms.READ, ZooDefs.Ids.ANYONE_ID_UNSAFE));
+  protected List<ACL> createNonSecurityACLsToAdd() {
+    List<ACL> ret = new ArrayList<ACL>();
+    ret.add(new ACL(ZooDefs.Perms.ALL, new Id("sasl", superUser)));
+    ret.add(new ACL(ZooDefs.Perms.READ, ZooDefs.Ids.ANYONE_ID_UNSAFE));
+    return ret;
+  }
 
-    if (result.isEmpty()) {
-      result = super.createGlobalACLsToAdd();
-    }
-    return result;
+  @Override
+  protected List<ACL> createSecurityACLsToAdd() {
+    List<ACL> ret = new ArrayList<ACL>();
+    ret.add(new ACL(ZooDefs.Perms.ALL, new Id("sasl", superUser)));
+    return ret;
   }
 }
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/SecurityAwareZkACLProvider.java b/solr/solrj/src/java/org/apache/solr/common/cloud/SecurityAwareZkACLProvider.java
new file mode 100644
index 0000000..1c74d94
--- /dev/null
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/SecurityAwareZkACLProvider.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.solr.common.cloud;
+
+import java.util.List;
+
+import org.apache.zookeeper.data.ACL;
+
+/**
+ * {@link ZkACLProvider} capable of returning a different set of
+ * {@link ACL}s for security-related znodes (default: subtree under /security)
+ * vs non-security-related znodes.
+ */
+public abstract class SecurityAwareZkACLProvider implements ZkACLProvider {
+  public static final String SECURITY_ZNODE_PATH = "/security";
+
+  private List<ACL> nonSecurityACLsToAdd;
+  private List<ACL> securityACLsToAdd;
+
+
+  @Override
+  public List<ACL> getACLsToAdd(String zNodePath) {
+    if (isSecurityZNodePath(zNodePath)) {
+      return getSecurityACLsToAdd();
+    } else {
+      return getNonSecurityACLsToAdd();
+    }
+  }
+
+  protected boolean isSecurityZNodePath(String zNodePath) {
+    if (zNodePath != null
+        && (zNodePath.equals(SECURITY_ZNODE_PATH) || zNodePath.startsWith(SECURITY_ZNODE_PATH + "/"))) {
+      return true;
+    }
+    return false;
+  }
+
+  /**
+   * @return Set of ACLs to return for non-security related znodes
+   */
+  protected abstract List<ACL> createNonSecurityACLsToAdd();
+
+  /**
+   * @return Set of ACLs to return security-related znodes
+   */
+  protected abstract List<ACL> createSecurityACLsToAdd();
+
+  private List<ACL> getNonSecurityACLsToAdd() {
+    if (nonSecurityACLsToAdd == null) {
+      synchronized (this) {
+        if (nonSecurityACLsToAdd == null) nonSecurityACLsToAdd = createNonSecurityACLsToAdd();
+      }
+    }
+    return nonSecurityACLsToAdd;
+  }
+
+  private List<ACL> getSecurityACLsToAdd() {
+    if (securityACLsToAdd == null) {
+      synchronized (this) {
+        if (securityACLsToAdd == null) securityACLsToAdd = createSecurityACLsToAdd();
+      }
+    }
+    return securityACLsToAdd;
+  }
+}
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java b/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java
index af6f9be..516b7b9 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/SolrZkClient.java
@@ -122,12 +122,12 @@
 
   public SolrZkClient(String zkServerAddress, int zkClientTimeout, int clientConnectTimeout,
       ZkClientConnectionStrategy strat, final OnReconnect onReconnect, BeforeReconnect beforeReconnect, ZkACLProvider zkACLProvider) {
-    this.zkClientConnectionStrategy = strat;
     this.zkServerAddress = zkServerAddress;
     
     if (strat == null) {
       strat = new DefaultConnectionStrategy();
     }
+    this.zkClientConnectionStrategy = strat;
 
     if (!strat.hasZkCredentialsToAddAutomatically()) {
       ZkCredentialsProvider zkCredentialsToAddAutomatically = createZkCredentialsToAddAutomatically();
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/VMParamsAllAndReadonlyDigestZkACLProvider.java b/solr/solrj/src/java/org/apache/solr/common/cloud/VMParamsAllAndReadonlyDigestZkACLProvider.java
index f6f491b..8866245 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/VMParamsAllAndReadonlyDigestZkACLProvider.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/VMParamsAllAndReadonlyDigestZkACLProvider.java
@@ -20,13 +20,14 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.solr.common.StringUtils;
 import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.data.Id;
 import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
 
-public class VMParamsAllAndReadonlyDigestZkACLProvider extends DefaultZkACLProvider {
+public class VMParamsAllAndReadonlyDigestZkACLProvider extends SecurityAwareZkACLProvider {
 
   public static final String DEFAULT_DIGEST_READONLY_USERNAME_VM_PARAM_NAME = "zkDigestReadonlyUsername";
   public static final String DEFAULT_DIGEST_READONLY_PASSWORD_VM_PARAM_NAME = "zkDigestReadonlyPassword";
@@ -53,29 +54,56 @@
     this.zkDigestReadonlyPasswordVMParamName = zkDigestReadonlyPasswordVMParamName;
   }
 
-
+  /**
+   * @return Set of ACLs to return for non-security related znodes
+   */
   @Override
-  protected List<ACL> createGlobalACLsToAdd() {
-    try {
+  protected List<ACL> createNonSecurityACLsToAdd() {
+    return createACLsToAdd(true);
+  }
+
+  /**
+   * @return Set of ACLs to return security-related znodes
+   */
+  @Override
+  protected List<ACL> createSecurityACLsToAdd() {
+    return createACLsToAdd(false);
+  }
+
+  protected List<ACL> createACLsToAdd(boolean includeReadOnly) {
+    String digestAllUsername = System.getProperty(zkDigestAllUsernameVMParamName);
+    String digestAllPassword = System.getProperty(zkDigestAllPasswordVMParamName);
+    String digestReadonlyUsername = System.getProperty(zkDigestReadonlyUsernameVMParamName);
+    String digestReadonlyPassword = System.getProperty(zkDigestReadonlyPasswordVMParamName);
+
+    return createACLsToAdd(includeReadOnly,
+        digestAllUsername, digestAllPassword,
+        digestReadonlyUsername, digestReadonlyPassword);
+  }
+
+  @VisibleForTesting
+  protected List<ACL> createACLsToAdd(boolean includeReadOnly,
+                                      String digestAllUsername, String digestAllPassword,
+                                      String digestReadonlyUsername, String digestReadonlyPassword) {
+
+      try {
       List<ACL> result = new ArrayList<ACL>();
   
       // Not to have to provide too much credentials and ACL information to the process it is assumed that you want "ALL"-acls
       // added to the user you are using to connect to ZK (if you are using VMParamsSingleSetCredentialsDigestZkCredentialsProvider)
-      String digestAllUsername = System.getProperty(zkDigestAllUsernameVMParamName);
-      String digestAllPassword = System.getProperty(zkDigestAllPasswordVMParamName);
       if (!StringUtils.isEmpty(digestAllUsername) && !StringUtils.isEmpty(digestAllPassword)) {
         result.add(new ACL(ZooDefs.Perms.ALL, new Id("digest", DigestAuthenticationProvider.generateDigest(digestAllUsername + ":" + digestAllPassword))));
       }
-  
-      // Besides that support for adding additional "READONLY"-acls for another user
-      String digestReadonlyUsername = System.getProperty(zkDigestReadonlyUsernameVMParamName);
-      String digestReadonlyPassword = System.getProperty(zkDigestReadonlyPasswordVMParamName);
-      if (!StringUtils.isEmpty(digestReadonlyUsername) && !StringUtils.isEmpty(digestReadonlyPassword)) {
-        result.add(new ACL(ZooDefs.Perms.READ, new Id("digest", DigestAuthenticationProvider.generateDigest(digestReadonlyUsername + ":" + digestReadonlyPassword))));
+
+      if (includeReadOnly) {
+        // Besides that support for adding additional "READONLY"-acls for another user
+        if (!StringUtils.isEmpty(digestReadonlyUsername) && !StringUtils.isEmpty(digestReadonlyPassword)) {
+          result.add(new ACL(ZooDefs.Perms.READ, new Id("digest", DigestAuthenticationProvider.generateDigest(digestReadonlyUsername + ":" + digestReadonlyPassword))));
+        }
       }
       
       if (result.isEmpty()) {
-        result = super.createGlobalACLsToAdd();
+        result = ZooDefs.Ids.OPEN_ACL_UNSAFE;
       }
       
       return result;
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkClientConnectionStrategy.java b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkClientConnectionStrategy.java
index 43a2c52..acc5abf 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkClientConnectionStrategy.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkClientConnectionStrategy.java
@@ -97,7 +97,9 @@
   public boolean hasZkCredentialsToAddAutomatically() {
     return zkCredentialsToAddAutomatically != null;
   }
-  
+
+  public ZkCredentialsProvider getZkCredentialsToAddAutomatically() { return zkCredentialsToAddAutomatically; }
+
   protected SolrZooKeeper createSolrZooKeeper(final String serverAddress, final int zkClientTimeout,
       final Watcher watcher) throws IOException {
     SolrZooKeeper result = new SolrZooKeeper(serverAddress, zkClientTimeout, watcher);
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/request/TestDelegationTokenRequest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/request/TestDelegationTokenRequest.java
new file mode 100644
index 0000000..47b8385
--- /dev/null
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/request/TestDelegationTokenRequest.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.solr.client.solrj.request;
+
+import org.apache.lucene.util.LuceneTestCase;
+
+import org.junit.Test;
+
+/**
+ * Test for DelegationTokenRequests
+ */
+public class TestDelegationTokenRequest extends LuceneTestCase {
+
+  @Test
+  public void testGetRequest() throws Exception {
+    // without renewer
+    DelegationTokenRequest.Get get = new DelegationTokenRequest.Get();
+    assertEquals("GETDELEGATIONTOKEN", get.getParams().get("op"));
+    assertNull(get.getParams().get("renewer"));
+
+
+    // with renewer
+    final String renewer = "test";
+    get = new DelegationTokenRequest.Get(renewer);
+    assertEquals("GETDELEGATIONTOKEN", get.getParams().get("op"));
+    assertEquals(renewer, get.getParams().get("renewer"));
+  }
+
+  @Test
+  public void testRenewRequest() throws Exception {
+    final String token = "testToken";
+    DelegationTokenRequest.Renew renew = new DelegationTokenRequest.Renew(token);
+    assertEquals("RENEWDELEGATIONTOKEN", renew.getParams().get("op"));
+    assertEquals(token, renew.getParams().get("token"));
+    assertTrue(renew.getQueryParams().contains("op"));
+    assertTrue(renew.getQueryParams().contains("token"));
+
+    // can handle null token
+    renew = new DelegationTokenRequest.Renew(null);
+    renew.getParams();
+  }
+
+  @Test
+  public void testCancelRequest() throws Exception {
+    final String token = "testToken";
+    DelegationTokenRequest.Cancel cancel = new DelegationTokenRequest.Cancel(token);
+    assertEquals("CANCELDELEGATIONTOKEN", cancel.getParams().get("op"));
+    assertEquals(token, cancel.getParams().get("token"));
+    assertTrue(cancel.getQueryParams().contains("op"));
+    assertTrue(cancel.getQueryParams().contains("token"));
+
+    // can handle null token
+    cancel = new DelegationTokenRequest.Cancel(null);
+    cancel.getParams();
+  }
+}
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/response/TestDelegationTokenResponse.java b/solr/solrj/src/test/org/apache/solr/client/solrj/response/TestDelegationTokenResponse.java
new file mode 100644
index 0000000..c376223
--- /dev/null
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/response/TestDelegationTokenResponse.java
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.response;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.io.IOUtils;
+
+import org.apache.lucene.util.LuceneTestCase;
+import org.apache.solr.client.solrj.ResponseParser;
+import org.apache.solr.client.solrj.request.DelegationTokenRequest;
+import org.apache.solr.common.SolrException;
+
+import org.junit.Test;
+
+import org.noggit.CharArr;
+import org.noggit.JSONWriter;
+
+public class TestDelegationTokenResponse extends LuceneTestCase {
+
+  private void delegationTokenResponse(DelegationTokenRequest request,
+      DelegationTokenResponse response, String responseBody) throws Exception {
+    ResponseParser parser = request.getResponseParser();
+    response.setResponse(parser.processResponse(
+      IOUtils.toInputStream(responseBody, "UTF-8"), "UTF-8"));
+  }
+
+  private String getNestedMapJson(String outerKey, String innerKey, Object innerValue) {
+    CharArr out = new CharArr();
+    JSONWriter w = new JSONWriter(out, 2);
+    Map<String, Object> innerMap = new HashMap<String, Object>();
+    innerMap.put(innerKey, innerValue);
+    Map<String, Map<String, Object>> outerMap = new HashMap<String, Map<String, Object>>();
+    outerMap.put(outerKey, innerMap);
+    w.write(outerMap);
+    return out.toString();
+  }
+
+  private String getMapJson(String key, Object value) {
+    CharArr out = new CharArr();
+    JSONWriter w = new JSONWriter(out, 2);
+    Map<String, Object> map = new HashMap<String, Object>();
+    map.put(key, value);
+    w.write(map);
+    return out.toString();
+  }
+
+  @Test
+  public void testGetResponse() throws Exception {
+    DelegationTokenRequest.Get getRequest = new DelegationTokenRequest.Get();
+    DelegationTokenResponse.Get getResponse = new DelegationTokenResponse.Get();
+
+    // not a map
+    try {
+      delegationTokenResponse(getRequest, getResponse, "");
+      getResponse.getDelegationToken();
+      fail("Expected SolrException");
+    } catch (SolrException se) {
+    }
+
+    // doesn't have Token outerMap
+    final String someToken = "someToken";
+    delegationTokenResponse(getRequest, getResponse, getNestedMapJson("NotToken", "urlString", someToken));
+    assertNull(getResponse.getDelegationToken());
+
+    // Token is not a map
+    try {
+      delegationTokenResponse(getRequest, getResponse, getMapJson("Token", someToken));
+      getResponse.getDelegationToken();
+      fail("Expected SolrException");
+    } catch (SolrException se) {
+    }
+
+    // doesn't have urlString
+    delegationTokenResponse(getRequest, getResponse, getNestedMapJson("Token", "notUrlString", someToken));
+    assertNull(getResponse.getDelegationToken());
+
+    // has Token + urlString
+    delegationTokenResponse(getRequest, getResponse, getNestedMapJson("Token", "urlString", someToken));
+    assertEquals(someToken, getResponse.getDelegationToken());
+  }
+
+  @Test
+  public void testRenewResponse() throws Exception {
+    DelegationTokenRequest.Renew renewRequest = new DelegationTokenRequest.Renew("token");
+    DelegationTokenResponse.Renew renewResponse = new DelegationTokenResponse.Renew();
+
+    // not a map
+    try {
+      delegationTokenResponse(renewRequest, renewResponse, "");
+      renewResponse.getExpirationTime();
+      fail("Expected SolrException");
+    } catch (SolrException se) {
+    }
+
+    // doesn't have long
+    delegationTokenResponse(renewRequest, renewResponse, getMapJson("notLong", "123"));
+    assertNull(renewResponse.getExpirationTime());
+
+    // long isn't valid
+    try {
+      delegationTokenResponse(renewRequest, renewResponse, getMapJson("long", "aaa"));
+      renewResponse.getExpirationTime();
+      fail("Expected SolrException");
+    } catch (SolrException se) {
+    }
+
+    // valid
+    Long expirationTime = new Long(Long.MAX_VALUE);
+    delegationTokenResponse(renewRequest, renewResponse,
+      getMapJson("long", expirationTime));
+    assertEquals(expirationTime, renewResponse.getExpirationTime());
+  }
+
+  @Test
+  public void testCancelResponse() throws Exception {
+    // expect empty response
+    DelegationTokenRequest.Cancel cancelRequest = new DelegationTokenRequest.Cancel("token");
+    DelegationTokenResponse.Cancel cancelResponse = new DelegationTokenResponse.Cancel();
+    delegationTokenResponse(cancelRequest, cancelResponse, "");
+  }
+}