kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jun...@apache.org
Subject [3/4] kafka git commit: KAFKA-1686; Implement SASL/Kerberos
Date Tue, 20 Oct 2015 21:13:43 GMT
http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosRule.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosRule.java b/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosRule.java
new file mode 100644
index 0000000..c1789db
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/kerberos/KerberosRule.java
@@ -0,0 +1,189 @@
+/**
+ * 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.kafka.common.security.kerberos;
+
+import java.io.IOException;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * An encoding of a rule for translating kerberos names.
+ */
+class KerberosRule {
+
+    /**
+     * A pattern that matches a string without '$' and then a single
+     * parameter with $n.
+     */
+    private static final Pattern PARAMETER_PATTERN = Pattern.compile("([^$]*)(\\$(\\d*))?");
+
+    /**
+     * A pattern that recognizes simple/non-simple names.
+     */
+    private static final Pattern NON_SIMPLE_PATTERN = Pattern.compile("[/@]");
+
+    private final String defaultRealm;
+    private final boolean isDefault;
+    private final int numOfComponents;
+    private final String format;
+    private final Pattern match;
+    private final Pattern fromPattern;
+    private final String toPattern;
+    private final boolean repeat;
+
+    KerberosRule(String defaultRealm) {
+        this.defaultRealm = defaultRealm;
+        isDefault = true;
+        numOfComponents = 0;
+        format = null;
+        match = null;
+        fromPattern = null;
+        toPattern = null;
+        repeat = false;
+    }
+
+    KerberosRule(String defaultRealm, int numOfComponents, String format, String match, String fromPattern,
+                 String toPattern, boolean repeat) {
+        this.defaultRealm = defaultRealm;
+        isDefault = false;
+        this.numOfComponents = numOfComponents;
+        this.format = format;
+        this.match = match == null ? null : Pattern.compile(match);
+        this.fromPattern =
+                fromPattern == null ? null : Pattern.compile(fromPattern);
+        this.toPattern = toPattern;
+        this.repeat = repeat;
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder buf = new StringBuilder();
+        if (isDefault) {
+            buf.append("DEFAULT");
+        } else {
+            buf.append("RULE:[");
+            buf.append(numOfComponents);
+            buf.append(':');
+            buf.append(format);
+            buf.append(']');
+            if (match != null) {
+                buf.append('(');
+                buf.append(match);
+                buf.append(')');
+            }
+            if (fromPattern != null) {
+                buf.append("s/");
+                buf.append(fromPattern);
+                buf.append('/');
+                buf.append(toPattern);
+                buf.append('/');
+                if (repeat) {
+                    buf.append('g');
+                }
+            }
+        }
+        return buf.toString();
+    }
+
+    /**
+     * Replace the numbered parameters of the form $n where n is from 1 to
+     * the length of params. Normal text is copied directly and $n is replaced
+     * by the corresponding parameter.
+     * @param format the string to replace parameters again
+     * @param params the list of parameters
+     * @return the generated string with the parameter references replaced.
+     * @throws KerberosNameParser.BadFormatString
+     */
+    static String replaceParameters(String format,
+                                    String[] params) throws KerberosNameParser.BadFormatString {
+        Matcher match = PARAMETER_PATTERN.matcher(format);
+        int start = 0;
+        StringBuilder result = new StringBuilder();
+        while (start < format.length() && match.find(start)) {
+            result.append(match.group(1));
+            String paramNum = match.group(3);
+            if (paramNum != null) {
+                try {
+                    int num = Integer.parseInt(paramNum);
+                    if (num < 0 || num > params.length) {
+                        throw new KerberosNameParser.BadFormatString("index " + num + " from " + format +
+                                " is outside of the valid range 0 to " +
+                                (params.length - 1));
+                    }
+                    result.append(params[num]);
+                } catch (NumberFormatException nfe) {
+                    throw new KerberosNameParser.BadFormatString("bad format in username mapping in " +
+                            paramNum, nfe);
+                }
+
+            }
+            start = match.end();
+        }
+        return result.toString();
+    }
+
+    /**
+     * Replace the matches of the from pattern in the base string with the value
+     * of the to string.
+     * @param base the string to transform
+     * @param from the pattern to look for in the base string
+     * @param to the string to replace matches of the pattern with
+     * @param repeat whether the substitution should be repeated
+     * @return
+     */
+    static String replaceSubstitution(String base, Pattern from, String to,
+                                      boolean repeat) {
+        Matcher match = from.matcher(base);
+        if (repeat) {
+            return match.replaceAll(to);
+        } else {
+            return match.replaceFirst(to);
+        }
+    }
+
+    /**
+     * Try to apply this rule to the given name represented as a parameter
+     * array.
+     * @param params first element is the realm, second and later elements are
+     *        are the components of the name "a/b@FOO" -> {"FOO", "a", "b"}
+     * @return the short name if this rule applies or null
+     * @throws IOException throws if something is wrong with the rules
+     */
+    String apply(String[] params) throws IOException {
+        String result = null;
+        if (isDefault) {
+            if (defaultRealm.equals(params[0])) {
+                result = params[1];
+            }
+        } else if (params.length - 1 == numOfComponents) {
+            String base = replaceParameters(format, params);
+            if (match == null || match.matcher(base).matches()) {
+                if (fromPattern == null) {
+                    result = base;
+                } else {
+                    result = replaceSubstitution(base, fromPattern, toPattern,  repeat);
+                }
+            }
+        }
+        if (result != null && NON_SIMPLE_PATTERN.matcher(result).find()) {
+            throw new NoMatchingRule("Non-simple name " + result + " after auth_to_local rule " + this);
+        }
+        return result;
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/clients/src/main/java/org/apache/kafka/common/security/kerberos/Login.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/kerberos/Login.java b/clients/src/main/java/org/apache/kafka/common/security/kerberos/Login.java
new file mode 100644
index 0000000..dd885e5
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/kerberos/Login.java
@@ -0,0 +1,389 @@
+/**
+ * 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.kafka.common.security.kerberos;
+
+import javax.security.auth.kerberos.KerberosPrincipal;
+import javax.security.auth.login.AppConfigurationEntry;
+import javax.security.auth.login.Configuration;
+import javax.security.auth.login.LoginContext;
+import javax.security.auth.login.LoginException;
+import javax.security.auth.callback.CallbackHandler;
+import javax.security.auth.kerberos.KerberosTicket;
+import javax.security.auth.Subject;
+
+import org.apache.kafka.common.security.authenticator.SaslClientAuthenticator.ClientCallbackHandler;
+import org.apache.kafka.common.security.JaasUtils;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.utils.Shell;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.SystemTime;
+
+import org.apache.kafka.common.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Date;
+import java.util.Random;
+import java.util.Set;
+import java.util.Map;
+
+/**
+ * This class is responsible for refreshing Kerberos credentials for
+ * logins for both Kafka client and server.
+ */
+public class Login {
+    private static final Logger log = LoggerFactory.getLogger(Login.class);
+
+    private static final Random RNG = new Random();
+
+    private final Thread t;
+    private final boolean isKrbTicket;
+    private final boolean isUsingTicketCache;
+
+    private final String loginContextName;
+    private final String principal;
+    private final Time time = new SystemTime();
+    private final CallbackHandler callbackHandler = new ClientCallbackHandler();
+
+    // LoginThread will sleep until 80% of time from last refresh to
+    // ticket's expiry has been reached, at which time it will wake
+    // and try to renew the ticket.
+    private final double ticketRenewWindowFactor;
+
+    /**
+     * Percentage of random jitter added to the renewal time
+     */
+    private final double ticketRenewJitter;
+
+    // Regardless of ticketRenewWindowFactor setting above and the ticket expiry time,
+    // thread will not sleep between refresh attempts any less than 1 minute (60*1000 milliseconds = 1 minute).
+    // Change the '1' to e.g. 5, to change this to 5 minutes.
+    private final long minTimeBeforeRelogin;
+
+    private final String kinitCmd;
+
+    private volatile Subject subject;
+
+    private LoginContext login;
+    private long lastLogin;
+
+    /**
+     * Login constructor. The constructor starts the thread used
+     * to periodically re-login to the Kerberos Ticket Granting Server.
+     * @param loginContextName
+     *               name of section in JAAS file that will be use to login.
+     *               Passed as first param to javax.security.auth.login.LoginContext().
+     * @param configs configure Login with the given key-value pairs.
+     * @throws javax.security.auth.login.LoginException
+     *               Thrown if authentication fails.
+     */
+    public Login(final String loginContextName, Map<String, ?> configs) throws LoginException {
+        this.loginContextName = loginContextName;
+        this.ticketRenewWindowFactor = (Double) configs.get(SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR);
+        this.ticketRenewJitter = (Double) configs.get(SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER);
+        this.minTimeBeforeRelogin = (Long) configs.get(SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN);
+        this.kinitCmd = (String) configs.get(SaslConfigs.SASL_KERBEROS_KINIT_CMD);
+
+        this.lastLogin = currentElapsedTime();
+        login = login(loginContextName);
+        subject = login.getSubject();
+        isKrbTicket = !subject.getPrivateCredentials(KerberosTicket.class).isEmpty();
+
+        AppConfigurationEntry[] entries = Configuration.getConfiguration().getAppConfigurationEntry(loginContextName);
+        if (entries.length == 0) {
+            isUsingTicketCache = false;
+            principal = null;
+        } else {
+            // there will only be a single entry
+            AppConfigurationEntry entry = entries[0];
+            if (entry.getOptions().get("useTicketCache") != null) {
+                String val = (String) entry.getOptions().get("useTicketCache");
+                isUsingTicketCache = val.equals("true");
+            } else
+                isUsingTicketCache = false;
+            if (entry.getOptions().get("principal") != null)
+                principal = (String) entry.getOptions().get("principal");
+            else
+                principal = null;
+        }
+
+        if (!isKrbTicket) {
+            log.debug("It is not a Kerberos ticket");
+            t = null;
+            // if no TGT, do not bother with ticket management.
+            return;
+        }
+        log.debug("It is a Kerberos ticket");
+
+        // Refresh the Ticket Granting Ticket (TGT) periodically. How often to refresh is determined by the
+        // TGT's existing expiry date and the configured minTimeBeforeRelogin. For testing and development,
+        // you can decrease the interval of expiration of tickets (for example, to 3 minutes) by running:
+        //  "modprinc -maxlife 3mins <principal>" in kadmin.
+        t = Utils.newThread("kafka-kerberos-refresh-thread", new Runnable() {
+            public void run() {
+                log.info("TGT refresh thread started.");
+                while (true) {  // renewal thread's main loop. if it exits from here, thread will exit.
+                    KerberosTicket tgt = getTGT();
+                    long now = currentWallTime();
+                    long nextRefresh;
+                    Date nextRefreshDate;
+                    if (tgt == null) {
+                        nextRefresh = now + minTimeBeforeRelogin;
+                        nextRefreshDate = new Date(nextRefresh);
+                        log.warn("No TGT found: will try again at {}", nextRefreshDate);
+                    } else {
+                        nextRefresh = getRefreshTime(tgt);
+                        long expiry = tgt.getEndTime().getTime();
+                        Date expiryDate = new Date(expiry);
+                        if (isUsingTicketCache && tgt.getRenewTill() != null && tgt.getRenewTill().getTime() >= expiry) {
+                            log.error("The TGT cannot be renewed beyond the next expiry date: {}." +
+                                    "This process will not be able to authenticate new SASL connections after that " +
+                                    "time (for example, it will not be able to authenticate a new connection with a Kafka " +
+                                    "Broker).  Ask your system administrator to either increase the " +
+                                    "'renew until' time by doing : 'modprinc -maxrenewlife {} ' within " +
+                                    "kadmin, or instead, to generate a keytab for {}. Because the TGT's " +
+                                    "expiry cannot be further extended by refreshing, exiting refresh thread now.",
+                                    expiryDate, principal, principal);
+                            return;
+                        }
+                        // determine how long to sleep from looking at ticket's expiry.
+                        // We should not allow the ticket to expire, but we should take into consideration
+                        // minTimeBeforeRelogin. Will not sleep less than minTimeBeforeRelogin, unless doing so
+                        // would cause ticket expiration.
+                        if ((nextRefresh > expiry) || (now + minTimeBeforeRelogin > expiry)) {
+                            // expiry is before next scheduled refresh).
+                            log.info("Refreshing now because expiry is before next scheduled refresh time.");
+                            nextRefresh = now;
+                        } else {
+                            if (nextRefresh < (now + minTimeBeforeRelogin)) {
+                                // next scheduled refresh is sooner than (now + MIN_TIME_BEFORE_LOGIN).
+                                Date until = new Date(nextRefresh);
+                                Date newUntil = new Date(now + minTimeBeforeRelogin);
+                                log.warn("TGT refresh thread time adjusted from {} to {} since the former is sooner " +
+                                        "than the minimum refresh interval ({} seconds) from now.",
+                                        until, newUntil, minTimeBeforeRelogin / 1000);
+                            }
+                            nextRefresh = Math.max(nextRefresh, now + minTimeBeforeRelogin);
+                        }
+                        nextRefreshDate = new Date(nextRefresh);
+                        if (nextRefresh > expiry) {
+                            log.error("Next refresh: {} is later than expiry {}. This may indicate a clock skew problem." +
+                                    "Check that this host and the KDC hosts' clocks are in sync. Exiting refresh thread.",
+                                    nextRefreshDate, expiryDate);
+                            return;
+                        }
+                    }
+                    if (now < nextRefresh) {
+                        Date until = new Date(nextRefresh);
+                        log.info("TGT refresh sleeping until: {}", until);
+                        try {
+                            Thread.sleep(nextRefresh - now);
+                        } catch (InterruptedException ie) {
+                            log.warn("TGT renewal thread has been interrupted and will exit.");
+                            return;
+                        }
+                    } else {
+                        log.error("NextRefresh: {} is in the past: exiting refresh thread. Check"
+                                + " clock sync between this host and KDC - (KDC's clock is likely ahead of this host)."
+                                + " Manual intervention will be required for this client to successfully authenticate."
+                                + " Exiting refresh thread.", nextRefreshDate);
+                        return;
+                    }
+                    if (isUsingTicketCache) {
+                        String kinitArgs = "-R";
+                        int retry = 1;
+                        while (retry >= 0) {
+                            try {
+                                log.debug("Running ticket cache refresh command: {} {}", kinitCmd, kinitArgs);
+                                Shell.execCommand(kinitCmd, kinitArgs);
+                                break;
+                            } catch (Exception e) {
+                                if (retry > 0) {
+                                    --retry;
+                                    // sleep for 10 seconds
+                                    try {
+                                        Thread.sleep(10 * 1000);
+                                    } catch (InterruptedException ie) {
+                                        log.error("Interrupted while renewing TGT, exiting Login thread");
+                                        return;
+                                    }
+                                } else {
+                                    log.warn("Could not renew TGT due to problem running shell command: '" + kinitCmd
+                                            + " " + kinitArgs + "'" + "; exception was: " + e + ". Exiting refresh thread.", e);
+                                    return;
+                                }
+                            }
+                        }
+                    }
+                    try {
+                        int retry = 1;
+                        while (retry >= 0) {
+                            try {
+                                reLogin();
+                                break;
+                            } catch (LoginException le) {
+                                if (retry > 0) {
+                                    --retry;
+                                    // sleep for 10 seconds.
+                                    try {
+                                        Thread.sleep(10 * 1000);
+                                    } catch (InterruptedException e) {
+                                        log.error("Interrupted during login retry after LoginException:", le);
+                                        throw le;
+                                    }
+                                } else {
+                                    log.error("Could not refresh TGT for principal: " + principal + ".", le);
+                                }
+                            }
+                        }
+                    } catch (LoginException le) {
+                        log.error("Failed to refresh TGT: refresh thread exiting now.", le);
+                        return;
+                    }
+                }
+            }
+        }, false);
+    }
+
+    public void startThreadIfNeeded() {
+        // thread object 't' will be null if a refresh thread is not needed.
+        if (t != null) {
+            t.start();
+        }
+    }
+
+    public void shutdown() {
+        if ((t != null) && (t.isAlive())) {
+            t.interrupt();
+            try {
+                t.join();
+            } catch (InterruptedException e) {
+                log.warn("Error while waiting for Login thread to shutdown: " + e, e);
+            }
+        }
+    }
+
+    public Subject subject() {
+        return subject;
+    }
+
+    private synchronized LoginContext login(final String loginContextName) throws LoginException {
+        String jaasConfigFile = System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM);
+        if (jaasConfigFile == null) {
+            throw new IllegalArgumentException("You must pass " + JaasUtils.JAVA_LOGIN_CONFIG_PARAM + " in secure mode.");
+        }
+
+        AppConfigurationEntry[] configEntries = Configuration.getConfiguration().getAppConfigurationEntry(loginContextName);
+        if (configEntries == null) {
+            // Forcing a reload of the configuration in case it's been overridden by third-party code.
+            // Without this, our tests fail sometimes depending on the order the tests are executed.
+            // Singletons are bad.
+            Configuration.setConfiguration(null);
+            configEntries = Configuration.getConfiguration().getAppConfigurationEntry(loginContextName);
+            if (configEntries == null) {
+                String errorMessage = "Could not find a '" + loginContextName + "' entry in `" + jaasConfigFile + "`.";
+                throw new IllegalArgumentException(errorMessage);
+            } else {
+                log.info("Found `" + loginContextName + "` in JAAS configuration after forced reload.");
+            }
+        }
+
+        LoginContext loginContext = new LoginContext(loginContextName, callbackHandler);
+        loginContext.login();
+        log.info("Successfully logged in.");
+        return loginContext;
+    }
+
+    private long getRefreshTime(KerberosTicket tgt) {
+        long start = tgt.getStartTime().getTime();
+        long expires = tgt.getEndTime().getTime();
+        log.info("TGT valid starting at: {}", tgt.getStartTime());
+        log.info("TGT expires: {}", tgt.getEndTime());
+        long proposedRefresh = start + (long) ((expires - start) *
+                (ticketRenewWindowFactor + (ticketRenewJitter * RNG.nextDouble())));
+
+        if (proposedRefresh > expires)
+            // proposedRefresh is too far in the future: it's after ticket expires: simply return now.
+            return currentWallTime();
+        else
+            return proposedRefresh;
+    }
+
+    private synchronized KerberosTicket getTGT() {
+        Set<KerberosTicket> tickets = subject.getPrivateCredentials(KerberosTicket.class);
+        for (KerberosTicket ticket : tickets) {
+            KerberosPrincipal server = ticket.getServer();
+            if (server.getName().equals("krbtgt/" + server.getRealm() + "@" + server.getRealm())) {
+                log.debug("Found TGT {}.", ticket);
+                return ticket;
+            }
+        }
+        return null;
+    }
+
+    private boolean hasSufficientTimeElapsed() {
+        long now = currentElapsedTime();
+        if (now - lastLogin < minTimeBeforeRelogin) {
+            log.warn("Not attempting to re-login since the last re-login was attempted less than {} seconds before.",
+                    minTimeBeforeRelogin / 1000);
+            return false;
+        }
+        return true;
+    }
+
+    /**
+     * Re-login a principal. This method assumes that {@link #login(String)} has happened already.
+     * @throws javax.security.auth.login.LoginException on a failure
+     */
+    private synchronized void reLogin()
+            throws LoginException {
+        if (!isKrbTicket) {
+            return;
+        }
+        if (login == null) {
+            throw new LoginException("Login must be done first");
+        }
+        if (!hasSufficientTimeElapsed()) {
+            return;
+        }
+        log.info("Initiating logout for {}", principal);
+        synchronized (Login.class) {
+            // register most recent relogin attempt
+            lastLogin = currentElapsedTime();
+            //clear up the kerberos state. But the tokens are not cleared! As per
+            //the Java kerberos login module code, only the kerberos credentials
+            //are cleared
+            login.logout();
+            //login and also update the subject field of this instance to
+            //have the new credentials (pass it to the LoginContext constructor)
+            login = new LoginContext(loginContextName, subject);
+            log.info("Initiating re-login for {}", principal);
+            login.login();
+        }
+    }
+
+    private long currentElapsedTime() {
+        return time.nanoseconds() / 1000000;
+    }
+
+    private long currentWallTime() {
+        return time.milliseconds();
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/clients/src/main/java/org/apache/kafka/common/security/kerberos/LoginManager.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/kerberos/LoginManager.java b/clients/src/main/java/org/apache/kafka/common/security/kerberos/LoginManager.java
new file mode 100644
index 0000000..18651c8
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/kerberos/LoginManager.java
@@ -0,0 +1,118 @@
+/**
+ * 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.kafka.common.security.kerberos;
+
+import javax.security.auth.Subject;
+import javax.security.auth.login.LoginException;
+import java.io.IOException;
+import java.util.EnumMap;
+import java.util.Map;
+
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.network.LoginType;
+import org.apache.kafka.common.security.JaasUtils;
+
+public class LoginManager {
+
+    private static final EnumMap<LoginType, LoginManager> CACHED_INSTANCES = new EnumMap(LoginType.class);
+
+    private final Login login;
+    private final String serviceName;
+    private final LoginType loginType;
+    private int refCount;
+
+    private LoginManager(LoginType loginType, Map<String, ?> configs) throws IOException, LoginException {
+        this.loginType = loginType;
+        String loginContext = loginType.contextName();
+        login = new Login(loginContext, configs);
+        this.serviceName = getServiceName(loginContext, configs);
+        login.startThreadIfNeeded();
+    }
+
+    private static String getServiceName(String loginContext, Map<String, ?> configs) throws IOException {
+        String jaasServiceName = JaasUtils.jaasConfig(loginContext, JaasUtils.SERVICE_NAME);
+        String configServiceName = (String) configs.get(SaslConfigs.SASL_KERBEROS_SERVICE_NAME);
+        if (jaasServiceName != null && configServiceName != null && jaasServiceName != configServiceName) {
+            String message = "Conflicting serviceName values found in JAAS and Kafka configs " +
+                "value in JAAS file " + jaasServiceName + ", value in Kafka config " + configServiceName;
+            throw new IllegalArgumentException(message);
+        }
+
+        if (jaasServiceName != null)
+            return jaasServiceName;
+        if (configServiceName != null)
+            return configServiceName;
+
+        throw new IllegalArgumentException("No serviceName defined in either JAAS or Kafka config");
+    }
+
+    /**
+     * Returns an instance of `LoginManager` and increases its reference count.
+     *
+     * `release()` should be invoked when the `LoginManager` is no longer needed. This method will try to reuse an
+     * existing `LoginManager` for the provided `mode` if available. However, it expects `configs` to be the same for
+     * every invocation and it will ignore them in the case where it's returning a cached instance of `LoginManager`.
+     *
+     * This is a bit ugly and it would be nicer if we could pass the `LoginManager` to `ChannelBuilders.create` and
+     * shut it down when the broker or clients are closed. It's straightforward to do the former, but it's more
+     * complicated to do the latter without making the consumer API more complex.
+     *
+     * @param loginType the type of the login context, it should be SERVER for the broker and CLIENT for the clients
+     *                  (i.e. consumer and producer)
+     * @param configs configuration as key/value pairs
+     */
+    public static final LoginManager acquireLoginManager(LoginType loginType, Map<String, ?> configs) throws IOException, LoginException {
+        synchronized (LoginManager.class) {
+            LoginManager loginManager = CACHED_INSTANCES.get(loginType);
+            if (loginManager == null) {
+                loginManager = new LoginManager(loginType, configs);
+                CACHED_INSTANCES.put(loginType, loginManager);
+            }
+            return loginManager.acquire();
+        }
+    }
+
+    public Subject subject() {
+        return login.subject();
+    }
+
+    public String serviceName() {
+        return serviceName;
+    }
+
+    private LoginManager acquire() {
+        ++refCount;
+        return this;
+    }
+
+    /**
+     * Decrease the reference count for this instance and release resources if it reaches 0.
+     */
+    public void release() {
+        synchronized (LoginManager.class) {
+            if (refCount == 0)
+                throw new IllegalStateException("release called on LoginManager with refCount == 0");
+            else if (refCount == 1) {
+                CACHED_INSTANCES.remove(loginType);
+                login.shutdown();
+            }
+            --refCount;
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/clients/src/main/java/org/apache/kafka/common/security/kerberos/NoMatchingRule.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/kerberos/NoMatchingRule.java b/clients/src/main/java/org/apache/kafka/common/security/kerberos/NoMatchingRule.java
new file mode 100644
index 0000000..6c2d267
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/security/kerberos/NoMatchingRule.java
@@ -0,0 +1,27 @@
+/**
+ * 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.kafka.common.security.kerberos;
+
+import java.io.IOException;
+
+public class NoMatchingRule extends IOException {
+    NoMatchingRule(String msg) {
+        super(msg);
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/clients/src/main/java/org/apache/kafka/common/security/ssl/SSLFactory.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/ssl/SSLFactory.java b/clients/src/main/java/org/apache/kafka/common/security/ssl/SSLFactory.java
index b291409..163b8c6 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/ssl/SSLFactory.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/ssl/SSLFactory.java
@@ -19,19 +19,19 @@ package org.apache.kafka.common.security.ssl;
 import org.apache.kafka.common.Configurable;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.config.SSLConfigs;
+import org.apache.kafka.common.network.Mode;
 
 import javax.net.ssl.*;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.security.GeneralSecurityException;
 import java.security.KeyStore;
+
 import java.util.List;
 import java.util.Map;
 
-
 public class SSLFactory implements Configurable {
 
-    public enum Mode { CLIENT, SERVER };
     private String protocol;
     private String provider;
     private String kmfAlgorithm;

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/clients/src/main/java/org/apache/kafka/common/utils/Shell.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Shell.java b/clients/src/main/java/org/apache/kafka/common/utils/Shell.java
new file mode 100644
index 0000000..f5db5c3
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/common/utils/Shell.java
@@ -0,0 +1,304 @@
+/**
+ * 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.kafka.common.utils;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A base class for running a Unix command.
+ *
+ * <code>Shell</code> can be used to run unix commands like <code>du</code> or
+ * <code>df</code>.
+ */
+abstract public class Shell {
+
+    private static final Logger LOG = LoggerFactory.getLogger(Shell.class);
+
+    /** Return an array containing the command name and its parameters */
+    protected abstract String[] execString();
+
+    /** Parse the execution result */
+    protected abstract void parseExecResult(BufferedReader lines) throws IOException;
+
+    private final long timeout;
+
+    private int exitCode;
+    private Process process; // sub process used to execute the command
+
+    /* If or not script finished executing */
+    private volatile AtomicBoolean completed;
+
+    /**
+     * @param timeout Specifies the time in milliseconds, after which the command will be killed. -1 means no timeout.
+     */
+    public Shell(long timeout) {
+        this.timeout = timeout;
+    }
+
+    /** get the exit code
+     * @return the exit code of the process
+     */
+    public int exitCode() {
+        return exitCode;
+    }
+
+    /** get the current sub-process executing the given command
+     * @return process executing the command
+     */
+    public Process process() {
+        return process;
+    }
+
+    protected void run() throws IOException {
+        exitCode = 0; // reset for next run
+        runCommand();
+    }
+
+    /** Run a command */
+    private void runCommand() throws IOException {
+        ProcessBuilder builder = new ProcessBuilder(execString());
+        Timer timeoutTimer = null;
+        completed = new AtomicBoolean(false);
+
+        process = builder.start();
+        if (timeout > -1) {
+            timeoutTimer = new Timer();
+            //One time scheduling.
+            timeoutTimer.schedule(new ShellTimeoutTimerTask(this), timeout);
+        }
+        final BufferedReader errReader = new BufferedReader(new InputStreamReader(process.getErrorStream()));
+        BufferedReader inReader = new BufferedReader(new InputStreamReader(process.getInputStream()));
+        final StringBuffer errMsg = new StringBuffer();
+
+        // read error and input streams as this would free up the buffers
+        // free the error stream buffer
+        Thread errThread = Utils.newThread("kafka-shell-thread", new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    String line = errReader.readLine();
+                    while ((line != null) && !Thread.currentThread().isInterrupted()) {
+                        errMsg.append(line);
+                        errMsg.append(System.getProperty("line.separator"));
+                        line = errReader.readLine();
+                    }
+                } catch (IOException ioe) {
+                    LOG.warn("Error reading the error stream", ioe);
+                }
+            }
+        }, false);
+        errThread.start();
+
+        try {
+            parseExecResult(inReader); // parse the output
+            // clear the input stream buffer
+            String line = null;
+            while (line != null) {
+                line = inReader.readLine();
+            }
+            // wait for the process to finish and check the exit code
+            exitCode = process.waitFor();
+            try {
+                // make sure that the error thread exits
+                errThread.join();
+            } catch (InterruptedException ie) {
+                LOG.warn("Interrupted while reading the error stream", ie);
+            }
+            completed.set(true);
+            //the timeout thread handling
+            //taken care in finally block
+            if (exitCode != 0) {
+                throw new ExitCodeException(exitCode, errMsg.toString());
+            }
+        } catch (InterruptedException ie) {
+            throw new IOException(ie.toString());
+        } finally {
+            if (timeoutTimer != null)
+                timeoutTimer.cancel();
+
+            // close the input stream
+            try {
+                inReader.close();
+            } catch (IOException ioe) {
+                LOG.warn("Error while closing the input stream", ioe);
+            }
+            if (!completed.get())
+                errThread.interrupt();
+
+            try {
+                errReader.close();
+            } catch (IOException ioe) {
+                LOG.warn("Error while closing the error stream", ioe);
+            }
+
+            process.destroy();
+        }
+    }
+
+
+    /**
+     * This is an IOException with exit code added.
+     */
+    @SuppressWarnings("serial")
+    public static class ExitCodeException extends IOException {
+        int exitCode;
+
+        public ExitCodeException(int exitCode, String message) {
+            super(message);
+            this.exitCode = exitCode;
+        }
+
+        public int getExitCode() {
+            return exitCode;
+        }
+    }
+
+    /**
+     * A simple shell command executor.
+     *
+     * <code>ShellCommandExecutor</code>should be used in cases where the output
+     * of the command needs no explicit parsing and where the command, working
+     * directory and the environment remains unchanged. The output of the command
+     * is stored as-is and is expected to be small.
+     */
+    public static class ShellCommandExecutor extends Shell {
+
+        private final String[] command;
+        private StringBuffer output;
+
+        /**
+         * Create a new instance of the ShellCommandExecutor to execute a command.
+         *
+         * @param execString The command to execute with arguments
+         * @param timeout Specifies the time in milliseconds, after which the
+         *                command will be killed. -1 means no timeout.
+         */
+
+        public ShellCommandExecutor(String[] execString, long timeout) {
+            super(timeout);
+            command = execString.clone();
+        }
+
+
+        /** Execute the shell command. */
+        public void execute() throws IOException {
+            this.run();
+        }
+
+        protected String[] execString() {
+            return command;
+        }
+
+        protected void parseExecResult(BufferedReader reader) throws IOException {
+            output = new StringBuffer();
+            char[] buf = new char[512];
+            int nRead;
+            while ((nRead = reader.read(buf, 0, buf.length)) > 0) {
+                output.append(buf, 0, nRead);
+            }
+        }
+
+        /** Get the output of the shell command.*/
+        public String output() {
+            return (output == null) ? "" : output.toString();
+        }
+
+        /**
+         * Returns the commands of this instance.
+         * Arguments with spaces in are presented with quotes round; other
+         * arguments are presented raw
+         *
+         * @return a string representation of the object.
+         */
+        public String toString() {
+            StringBuilder builder = new StringBuilder();
+            String[] args = execString();
+            for (String s : args) {
+                if (s.indexOf(' ') >= 0) {
+                    builder.append('"').append(s).append('"');
+                } else {
+                    builder.append(s);
+                }
+                builder.append(' ');
+            }
+            return builder.toString();
+        }
+    }
+
+    /**
+     * Static method to execute a shell command.
+     * Covers most of the simple cases without requiring the user to implement
+     * the <code>Shell</code> interface.
+     * @param cmd shell command to execute.
+     * @return the output of the executed command.
+     */
+    public static String execCommand(String ... cmd) throws IOException {
+        return execCommand(cmd, -1);
+    }
+
+    /**
+     * Static method to execute a shell command.
+     * Covers most of the simple cases without requiring the user to implement
+     * the <code>Shell</code> interface.
+     * @param cmd shell command to execute.
+     * @param timeout time in milliseconds after which script should be killed. -1 means no timeout.
+     * @return the output of the executed command.
+     */
+    public static String execCommand(String[] cmd, long timeout) throws IOException {
+        ShellCommandExecutor exec = new ShellCommandExecutor(cmd, timeout);
+        exec.execute();
+        return exec.output();
+    }
+
+    /**
+     * Timer which is used to timeout scripts spawned off by shell.
+     */
+    private static class ShellTimeoutTimerTask extends TimerTask {
+
+        private final Shell shell;
+
+        public ShellTimeoutTimerTask(Shell shell) {
+            this.shell = shell;
+        }
+
+        @Override
+        public void run() {
+            Process p = shell.process();
+            try {
+                p.exitValue();
+            } catch (Exception e) {
+                //Process has not terminated.
+                //So check if it has completed
+                //if not just destroy it.
+                if (p != null && !shell.completed.get()) {
+                    p.destroy();
+                }
+            }
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/clients/src/main/java/org/apache/kafka/common/utils/Time.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Time.java b/clients/src/main/java/org/apache/kafka/common/utils/Time.java
index 66c44de..b2fad7f 100644
--- a/clients/src/main/java/org/apache/kafka/common/utils/Time.java
+++ b/clients/src/main/java/org/apache/kafka/common/utils/Time.java
@@ -24,16 +24,16 @@ public interface Time {
     /**
      * The current time in milliseconds
      */
-    public long milliseconds();
+    long milliseconds();
 
     /**
      * The current time in nanoseconds
      */
-    public long nanoseconds();
+    long nanoseconds();
 
     /**
      * Sleep for the given number of milliseconds
      */
-    public void sleep(long ms);
+    void sleep(long ms);
 
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/clients/src/test/java/org/apache/kafka/common/network/EchoServer.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/network/EchoServer.java b/clients/src/test/java/org/apache/kafka/common/network/EchoServer.java
index f13c21a..4a6d304 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/EchoServer.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/EchoServer.java
@@ -45,7 +45,7 @@ class EchoServer extends Thread {
         this.protocol =  configs.containsKey("security.protocol") ?
             SecurityProtocol.valueOf((String) configs.get("security.protocol")) : SecurityProtocol.PLAINTEXT;
         if (protocol == SecurityProtocol.SSL) {
-            this.sslFactory = new SSLFactory(SSLFactory.Mode.SERVER);
+            this.sslFactory = new SSLFactory(Mode.SERVER);
             this.sslFactory.configure(configs);
             SSLContext sslContext = this.sslFactory.sslContext();
             this.serverSocket = sslContext.getServerSocketFactory().createServerSocket(0);

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/clients/src/test/java/org/apache/kafka/common/network/SSLSelectorTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SSLSelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/SSLSelectorTest.java
index c60053f..6475ff0 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SSLSelectorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SSLSelectorTest.java
@@ -22,7 +22,6 @@ import java.net.InetSocketAddress;
 
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.config.SSLConfigs;
-import org.apache.kafka.common.security.ssl.SSLFactory;
 import org.apache.kafka.common.utils.MockTime;
 import org.apache.kafka.test.TestSSLUtils;
 import org.junit.After;
@@ -40,15 +39,15 @@ public class SSLSelectorTest extends SelectorTest {
     public void setup() throws Exception {
         File trustStoreFile = File.createTempFile("truststore", ".jks");
 
-        Map<String, Object> sslServerConfigs = TestSSLUtils.createSSLConfig(false, true, SSLFactory.Mode.SERVER, trustStoreFile, "server");
+        Map<String, Object> sslServerConfigs = TestSSLUtils.createSSLConfig(false, true, Mode.SERVER, trustStoreFile, "server");
         sslServerConfigs.put(SSLConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, Class.forName(SSLConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS));
         this.server = new EchoServer(sslServerConfigs);
         this.server.start();
         this.time = new MockTime();
-        Map<String, Object> sslClientConfigs = TestSSLUtils.createSSLConfig(false, false, SSLFactory.Mode.SERVER, trustStoreFile, "client");
+        Map<String, Object> sslClientConfigs = TestSSLUtils.createSSLConfig(false, false, Mode.SERVER, trustStoreFile, "client");
         sslClientConfigs.put(SSLConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, Class.forName(SSLConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS));
 
-        this.channelBuilder = new SSLChannelBuilder(SSLFactory.Mode.CLIENT);
+        this.channelBuilder = new SSLChannelBuilder(Mode.CLIENT);
         this.channelBuilder.configure(sslClientConfigs);
         this.metrics = new Metrics();
         this.selector = new Selector(5000, metrics, time, "MetricGroup", new LinkedHashMap<String, String>(), channelBuilder);

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/clients/src/test/java/org/apache/kafka/common/network/SSLTransportLayerTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SSLTransportLayerTest.java b/clients/src/test/java/org/apache/kafka/common/network/SSLTransportLayerTest.java
index 6993f52..987f4bb 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SSLTransportLayerTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SSLTransportLayerTest.java
@@ -73,7 +73,7 @@ public class SSLTransportLayerTest {
         sslServerConfigs = serverCertStores.getTrustingConfig(clientCertStores);
         sslClientConfigs = clientCertStores.getTrustingConfig(serverCertStores);
 
-        this.channelBuilder = new SSLChannelBuilder(SSLFactory.Mode.CLIENT);
+        this.channelBuilder = new SSLChannelBuilder(Mode.CLIENT);
         this.channelBuilder.configure(sslClientConfigs);
         this.selector = new Selector(5000, new Metrics(), new MockTime(), "MetricGroup", new LinkedHashMap<String, String>(), channelBuilder);
     }
@@ -270,7 +270,7 @@ public class SSLTransportLayerTest {
      */
     @Test
     public void testInvalidTruststorePassword() throws Exception {
-        SSLChannelBuilder channelBuilder = new SSLChannelBuilder(SSLFactory.Mode.CLIENT);
+        SSLChannelBuilder channelBuilder = new SSLChannelBuilder(Mode.CLIENT);
         try {
             sslClientConfigs.put(SSLConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "invalid");
             channelBuilder.configure(sslClientConfigs);
@@ -285,7 +285,7 @@ public class SSLTransportLayerTest {
      */
     @Test
     public void testInvalidKeystorePassword() throws Exception {
-        SSLChannelBuilder channelBuilder = new SSLChannelBuilder(SSLFactory.Mode.CLIENT);
+        SSLChannelBuilder channelBuilder = new SSLChannelBuilder(Mode.CLIENT);
         try {
             sslClientConfigs.put(SSLConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, "invalid");
             channelBuilder.configure(sslClientConfigs);
@@ -437,7 +437,7 @@ public class SSLTransportLayerTest {
 
     private void createSelector(Map<String, Object> sslClientConfigs, final Integer netReadBufSize, final Integer netWriteBufSize, final Integer appBufSize) {
         
-        this.channelBuilder = new SSLChannelBuilder(SSLFactory.Mode.CLIENT) {
+        this.channelBuilder = new SSLChannelBuilder(Mode.CLIENT) {
 
             @Override
             protected SSLTransportLayer buildTransportLayer(SSLFactory sslFactory, String id, SelectionKey key) throws IOException {
@@ -461,7 +461,7 @@ public class SSLTransportLayerTest {
         
         CertStores(boolean server) throws Exception {
             String name = server ? "server" : "client";
-            SSLFactory.Mode mode = server ? SSLFactory.Mode.SERVER : SSLFactory.Mode.CLIENT;
+            Mode mode = server ? Mode.SERVER : Mode.CLIENT;
             File truststoreFile = File.createTempFile(name + "TS", ".jks");
             sslConfig = TestSSLUtils.createSSLConfig(!server, true, mode, truststoreFile, name);
             sslConfig.put(SSLConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, Class.forName(SSLConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS));
@@ -549,7 +549,7 @@ public class SSLTransportLayerTest {
         private final ConcurrentLinkedQueue<NetworkSend> inflightSends = new ConcurrentLinkedQueue<NetworkSend>();
 
         public SSLEchoServer(Map<String, ?> configs, String serverHost) throws Exception {
-            this.sslFactory = new SSLFactory(SSLFactory.Mode.SERVER);
+            this.sslFactory = new SSLFactory(Mode.SERVER);
             this.sslFactory.configure(configs);
             serverSocketChannel = ServerSocketChannel.open();
             serverSocketChannel.configureBlocking(false);
@@ -557,7 +557,7 @@ public class SSLTransportLayerTest {
             this.port = serverSocketChannel.socket().getLocalPort();
             this.socketChannels = Collections.synchronizedList(new ArrayList<SocketChannel>());
             this.newChannels = Collections.synchronizedList(new ArrayList<SocketChannel>());
-            SSLChannelBuilder channelBuilder = new SSLChannelBuilder(SSLFactory.Mode.SERVER);
+            SSLChannelBuilder channelBuilder = new SSLChannelBuilder(Mode.SERVER);
             channelBuilder.configure(sslServerConfigs);
             this.selector = new Selector(5000, new Metrics(), new MockTime(), "MetricGroup", new LinkedHashMap<String, String>(), channelBuilder);
             setName("echoserver");

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/clients/src/test/java/org/apache/kafka/common/security/kerberos/KerberosNameTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/kerberos/KerberosNameTest.java b/clients/src/test/java/org/apache/kafka/common/security/kerberos/KerberosNameTest.java
new file mode 100644
index 0000000..9781f6d
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/common/security/kerberos/KerberosNameTest.java
@@ -0,0 +1,59 @@
+/**
+ * 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.kafka.common.security.kerberos;
+
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+public class KerberosNameTest {
+
+    @Test
+    public void testParse() throws IOException {
+        List<String> rules = new ArrayList<>(Arrays.asList(
+            "RULE:[1:$1](App\\..*)s/App\\.(.*)/$1/g",
+            "RULE:[2:$1](App\\..*)s/App\\.(.*)/$1/g",
+            "DEFAULT"
+        ));
+        KerberosNameParser parser = new KerberosNameParser("REALM.COM", rules);
+
+        KerberosName name = parser.parse("App.service-name/example.com@REALM.COM");
+        assertEquals("App.service-name", name.serviceName());
+        assertEquals("example.com", name.hostName());
+        assertEquals("REALM.COM", name.realm());
+        assertEquals("service-name", name.shortName());
+
+        name = parser.parse("App.service-name@REALM.COM");
+        assertEquals("App.service-name", name.serviceName());
+        assertNull(name.hostName());
+        assertEquals("REALM.COM", name.realm());
+        assertEquals("service-name", name.shortName());
+
+        name = parser.parse("user/host@REALM.COM");
+        assertEquals("user", name.serviceName());
+        assertEquals("host", name.hostName());
+        assertEquals("REALM.COM", name.realm());
+        assertEquals("user", name.shortName());
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/clients/src/test/java/org/apache/kafka/common/security/ssl/SSLFactoryTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/ssl/SSLFactoryTest.java b/clients/src/test/java/org/apache/kafka/common/security/ssl/SSLFactoryTest.java
index 0aec666..e90ec2b 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/ssl/SSLFactoryTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/ssl/SSLFactoryTest.java
@@ -18,6 +18,7 @@ import java.io.File;
 import java.util.Map;
 
 import org.apache.kafka.test.TestSSLUtils;
+import org.apache.kafka.common.network.Mode;
 
 import org.junit.Test;
 import static org.junit.Assert.assertNotNull;
@@ -35,8 +36,8 @@ public class SSLFactoryTest {
     @Test
     public void testSSLFactoryConfiguration() throws Exception {
         File trustStoreFile = File.createTempFile("truststore", ".jks");
-        Map<String, Object> serverSSLConfig = TestSSLUtils.createSSLConfig(false, true, SSLFactory.Mode.SERVER, trustStoreFile, "server");
-        SSLFactory sslFactory = new SSLFactory(SSLFactory.Mode.SERVER);
+        Map<String, Object> serverSSLConfig = TestSSLUtils.createSSLConfig(false, true, Mode.SERVER, trustStoreFile, "server");
+        SSLFactory sslFactory = new SSLFactory(Mode.SERVER);
         sslFactory.configure(serverSSLConfig);
         //host and port are hints
         SSLEngine engine = sslFactory.createSSLEngine("localhost", 0);
@@ -49,8 +50,8 @@ public class SSLFactoryTest {
     @Test
     public void testClientMode() throws Exception {
         File trustStoreFile = File.createTempFile("truststore", ".jks");
-        Map<String, Object> clientSSLConfig = TestSSLUtils.createSSLConfig(false, true, SSLFactory.Mode.CLIENT, trustStoreFile, "client");
-        SSLFactory sslFactory = new SSLFactory(SSLFactory.Mode.CLIENT);
+        Map<String, Object> clientSSLConfig = TestSSLUtils.createSSLConfig(false, true, Mode.CLIENT, trustStoreFile, "client");
+        SSLFactory sslFactory = new SSLFactory(Mode.CLIENT);
         sslFactory.configure(clientSSLConfig);
         //host and port are hints
         SSLEngine engine = sslFactory.createSSLEngine("localhost", 0);

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/clients/src/test/java/org/apache/kafka/common/utils/MockTime.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/utils/MockTime.java b/clients/src/test/java/org/apache/kafka/common/utils/MockTime.java
index eb7fcf0..387e48f 100644
--- a/clients/src/test/java/org/apache/kafka/common/utils/MockTime.java
+++ b/clients/src/test/java/org/apache/kafka/common/utils/MockTime.java
@@ -3,9 +3,9 @@
  * 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.

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/clients/src/test/java/org/apache/kafka/test/TestSSLUtils.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/test/TestSSLUtils.java b/clients/src/test/java/org/apache/kafka/test/TestSSLUtils.java
index c01cf37..b231692 100644
--- a/clients/src/test/java/org/apache/kafka/test/TestSSLUtils.java
+++ b/clients/src/test/java/org/apache/kafka/test/TestSSLUtils.java
@@ -18,7 +18,7 @@
 package org.apache.kafka.test;
 
 import org.apache.kafka.common.config.SSLConfigs;
-import org.apache.kafka.common.security.ssl.SSLFactory;
+import org.apache.kafka.common.network.Mode;
 import org.apache.kafka.clients.CommonClientConfigs;
 
 import java.io.File;
@@ -177,13 +177,13 @@ public class TestSSLUtils {
         return certs;
     }
 
-    public static Map<String, Object> createSSLConfig(SSLFactory.Mode mode, File keyStoreFile, String password, String keyPassword,
+    public static Map<String, Object> createSSLConfig(Mode mode, File keyStoreFile, String password, String keyPassword,
                                                       File trustStoreFile, String trustStorePassword) {
         Map<String, Object> sslConfigs = new HashMap<String, Object>();
         sslConfigs.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL"); // kafka security protocol
         sslConfigs.put(SSLConfigs.SSL_PROTOCOL_CONFIG, "TLSv1.2"); // protocol to create SSLContext
 
-        if (mode == SSLFactory.Mode.SERVER || (mode == SSLFactory.Mode.CLIENT && keyStoreFile != null)) {
+        if (mode == Mode.SERVER || (mode == Mode.CLIENT && keyStoreFile != null)) {
             sslConfigs.put(SSLConfigs.SSL_KEYSTORE_LOCATION_CONFIG, keyStoreFile.getPath());
             sslConfigs.put(SSLConfigs.SSL_KEYSTORE_TYPE_CONFIG, "JKS");
             sslConfigs.put(SSLConfigs.SSL_KEYMANAGER_ALGORITHM_CONFIG, TrustManagerFactory.getDefaultAlgorithm());
@@ -203,13 +203,13 @@ public class TestSSLUtils {
         return sslConfigs;
     }
 
-    public static  Map<String, Object> createSSLConfig(boolean useClientCert, boolean trustStore, SSLFactory.Mode mode, File trustStoreFile, String certAlias)
+    public static  Map<String, Object> createSSLConfig(boolean useClientCert, boolean trustStore, Mode mode, File trustStoreFile, String certAlias)
         throws IOException, GeneralSecurityException {
         Map<String, X509Certificate> certs = new HashMap<String, X509Certificate>();
         File keyStoreFile;
         String password;
 
-        if (mode == SSLFactory.Mode.SERVER)
+        if (mode == Mode.SERVER)
             password = "ServerPassword";
         else
             password = "ClientPassword";

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
index c2076a2..3756822 100755
--- a/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
+++ b/core/src/main/scala/kafka/controller/ControllerChannelManager.scala
@@ -21,10 +21,9 @@ import kafka.utils._
 import org.apache.kafka.clients.{ClientResponse, ClientRequest, ManualMetadataUpdater, NetworkClient}
 import org.apache.kafka.common.{TopicPartition, Node}
 import org.apache.kafka.common.metrics.Metrics
-import org.apache.kafka.common.network.{Selectable, ChannelBuilders, Selector, NetworkReceive}
+import org.apache.kafka.common.network.{LoginType, Selectable, ChannelBuilders, Selector, NetworkReceive, Mode}
 import org.apache.kafka.common.protocol.{SecurityProtocol, ApiKeys}
 import org.apache.kafka.common.requests._
-import org.apache.kafka.common.security.ssl.SSLFactory
 import org.apache.kafka.common.utils.Time
 import collection.mutable.HashMap
 import kafka.cluster.Broker
@@ -97,7 +96,7 @@ class ControllerChannelManager(controllerContext: ControllerContext, config: Kaf
         "controller-channel",
         Map("broker-id" -> broker.id.toString).asJava,
         false,
-        ChannelBuilders.create(config.interBrokerSecurityProtocol, SSLFactory.Mode.CLIENT, config.channelConfigs)
+        ChannelBuilders.create(config.interBrokerSecurityProtocol, Mode.CLIENT, LoginType.SERVER, config.channelConfigs)
       )
       new NetworkClient(
         selector,

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/core/src/main/scala/kafka/network/SocketServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala
index 41a3705..1066fbe 100644
--- a/core/src/main/scala/kafka/network/SocketServer.scala
+++ b/core/src/main/scala/kafka/network/SocketServer.scala
@@ -33,9 +33,8 @@ import kafka.server.KafkaConfig
 import kafka.utils._
 import org.apache.kafka.common.MetricName
 import org.apache.kafka.common.metrics._
-import org.apache.kafka.common.network.{Selector => KSelector, ChannelBuilders, InvalidReceiveException}
+import org.apache.kafka.common.network.{Selector => KSelector, LoginType, Mode, ChannelBuilders, InvalidReceiveException}
 import org.apache.kafka.common.security.auth.KafkaPrincipal
-import org.apache.kafka.common.security.ssl.SSLFactory
 import org.apache.kafka.common.protocol.SecurityProtocol
 import org.apache.kafka.common.protocol.types.SchemaException
 import org.apache.kafka.common.utils.{Time, Utils}
@@ -378,7 +377,7 @@ private[kafka] class Processor(val id: Int,
 
   private val newConnections = new ConcurrentLinkedQueue[SocketChannel]()
   private val inflightResponses = mutable.Map[String, RequestChannel.Response]()
-  private val channelBuilder = ChannelBuilders.create(protocol, SSLFactory.Mode.SERVER, channelConfigs)
+  private val channelBuilder = ChannelBuilders.create(protocol, Mode.SERVER, LoginType.SERVER, channelConfigs)
   private val metricTags = new util.HashMap[String, String]()
   metricTags.put("networkProcessor", id.toString)
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/core/src/main/scala/kafka/server/KafkaConfig.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 913d49b..194ee9c 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -26,6 +26,8 @@ import kafka.consumer.ConsumerConfig
 import kafka.message.{BrokerCompressionCodec, CompressionCodec, Message, MessageSet}
 import kafka.utils.CoreUtils
 import org.apache.kafka.clients.CommonClientConfigs
+import org.apache.kafka.common.config.SaslConfigs
+
 import org.apache.kafka.common.config.{AbstractConfig, ConfigDef, SSLConfigs}
 import org.apache.kafka.common.metrics.MetricsReporter
 import org.apache.kafka.common.protocol.SecurityProtocol
@@ -33,7 +35,6 @@ import org.apache.kafka.common.security.auth.PrincipalBuilder
 
 import scala.collection.{Map, immutable}
 
-
 object Defaults {
   /** ********* Zookeeper Configuration ***********/
   val ZkSessionTimeoutMs = 6000
@@ -168,6 +169,14 @@ object Defaults {
   val SSLClientAuthNone = "none"
   val SSLClientAuth = SSLClientAuthNone
   val SSLCipherSuites = ""
+
+  /** ********* Sasl configuration ***********/
+  val SaslKerberosKinitCmd = SaslConfigs.DEFAULT_KERBEROS_KINIT_CMD
+  val SaslKerberosTicketRenewWindowFactor = SaslConfigs.DEFAULT_KERBEROS_TICKET_RENEW_WINDOW_FACTOR
+  val SaslKerberosTicketRenewJitter = SaslConfigs.DEFAULT_KERBEROS_TICKET_RENEW_JITTER
+  val SaslKerberosMinTimeBeforeRelogin = SaslConfigs.DEFAULT_KERBEROS_MIN_TIME_BEFORE_RELOGIN
+  val AuthToLocal = SaslConfigs.DEFAULT_AUTH_TO_LOCAL
+
 }
 
 object KafkaConfig {
@@ -316,6 +325,13 @@ object KafkaConfig {
   val SSLEndpointIdentificationAlgorithmProp = SSLConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG
   val SSLClientAuthProp = SSLConfigs.SSL_CLIENT_AUTH_CONFIG
 
+  /** ********* SASL Configuration ****************/
+  val SaslKerberosServiceNameProp = SaslConfigs.SASL_KERBEROS_SERVICE_NAME
+  val SaslKerberosKinitCmdProp = SaslConfigs.SASL_KERBEROS_KINIT_CMD
+  val SaslKerberosTicketRenewWindowFactorProp = SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR
+  val SaslKerberosTicketRenewJitterProp = SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER
+  val SaslKerberosMinTimeBeforeReloginProp = SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN
+  val AuthToLocalProp = SaslConfigs.AUTH_TO_LOCAL
 
   /* Documentation */
   /** ********* Zookeeper Configuration ***********/
@@ -487,6 +503,14 @@ object KafkaConfig {
   val SSLEndpointIdentificationAlgorithmDoc = SSLConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_DOC
   val SSLClientAuthDoc = SSLConfigs.SSL_CLIENT_AUTH_DOC
 
+  /** ********* Sasl Configuration ****************/
+  val SaslKerberosServiceNameDoc = SaslConfigs.SASL_KERBEROS_SERVICE_NAME_DOC
+  val SaslKerberosKinitCmdDoc = SaslConfigs.SASL_KERBEROS_KINIT_CMD_DOC
+  val SaslKerberosTicketRenewWindowFactorDoc = SaslConfigs.SASL_KERBEROS_TICKET_RENEW_WINDOW_FACTOR_DOC
+  val SaslKerberosTicketRenewJitterDoc = SaslConfigs.SASL_KERBEROS_TICKET_RENEW_JITTER_DOC
+  val SaslKerberosMinTimeBeforeReloginDoc = SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN_DOC
+  val AuthToLocalDoc = SaslConfigs.AUTH_TO_LOCAL_DOC
+
   private val configDef = {
     import ConfigDef.Importance._
     import ConfigDef.Range._
@@ -642,6 +666,15 @@ object KafkaConfig {
       .define(SSLTrustManagerAlgorithmProp, STRING, Defaults.SSLTrustManagerAlgorithm, MEDIUM, SSLTrustManagerAlgorithmDoc)
       .define(SSLClientAuthProp, STRING, Defaults.SSLClientAuth, in(Defaults.SSLClientAuthRequired, Defaults.SSLClientAuthRequested, Defaults.SSLClientAuthNone), MEDIUM, SSLClientAuthDoc)
       .define(SSLCipherSuitesProp, LIST, Defaults.SSLCipherSuites, MEDIUM, SSLCipherSuitesDoc)
+
+      /** ********* Sasl Configuration ****************/
+      .define(SaslKerberosServiceNameProp, STRING, MEDIUM, SaslKerberosServiceNameDoc, false)
+      .define(SaslKerberosKinitCmdProp, STRING, Defaults.SaslKerberosKinitCmd, MEDIUM, SaslKerberosKinitCmdDoc)
+      .define(SaslKerberosTicketRenewWindowFactorProp, DOUBLE, Defaults.SaslKerberosTicketRenewWindowFactor, MEDIUM, SaslKerberosTicketRenewWindowFactorDoc)
+      .define(SaslKerberosTicketRenewJitterProp, DOUBLE, Defaults.SaslKerberosTicketRenewJitter, MEDIUM, SaslKerberosTicketRenewJitterDoc)
+      .define(SaslKerberosMinTimeBeforeReloginProp, LONG, Defaults.SaslKerberosMinTimeBeforeRelogin, MEDIUM, SaslKerberosMinTimeBeforeReloginDoc)
+      .define(AuthToLocalProp, LIST, Defaults.AuthToLocal, MEDIUM, AuthToLocalDoc)
+
   }
 
   def configNames() = {
@@ -802,6 +835,14 @@ case class KafkaConfig (props: java.util.Map[_, _]) extends AbstractConfig(Kafka
   val sslClientAuth = getString(KafkaConfig.SSLClientAuthProp)
   val sslCipher = getList(KafkaConfig.SSLCipherSuitesProp)
 
+  /** ********* Sasl Configuration **************/
+  val saslKerberosServiceName = getString(KafkaConfig.SaslKerberosServiceNameProp)
+  val saslKerberosKinitCmd = getString(KafkaConfig.SaslKerberosKinitCmdProp)
+  val saslKerberosTicketRenewWindowFactor = getDouble(KafkaConfig.SaslKerberosTicketRenewWindowFactorProp)
+  val saslKerberosTicketRenewJitter = getDouble(KafkaConfig.SaslKerberosTicketRenewJitterProp)
+  val saslKerberosMinTimeBeforeRelogin = getLong(KafkaConfig.SaslKerberosMinTimeBeforeReloginProp)
+  val authToLocal = getList(KafkaConfig.AuthToLocalProp)
+
   /** ********* Quota Configuration **************/
   val producerQuotaBytesPerSecondDefault = getLong(KafkaConfig.ProducerQuotaBytesPerSecondDefaultProp)
   val consumerQuotaBytesPerSecondDefault = getLong(KafkaConfig.ConsumerQuotaBytesPerSecondDefaultProp)
@@ -823,7 +864,7 @@ case class KafkaConfig (props: java.util.Map[_, _]) extends AbstractConfig(Kafka
     val millis: java.lang.Long =
       Option(getLong(KafkaConfig.LogRetentionTimeMillisProp)).getOrElse(
         Option(getInt(KafkaConfig.LogRetentionTimeMinutesProp)) match {
-          case Some(mins) =>  millisInMinute * mins
+          case Some(mins) => millisInMinute * mins
           case None => getInt(KafkaConfig.LogRetentionTimeHoursProp) * millisInHour
         })
 
@@ -927,20 +968,30 @@ case class KafkaConfig (props: java.util.Map[_, _]) extends AbstractConfig(Kafka
   def channelConfigs: java.util.Map[String, Object] = {
     val channelConfigs = new java.util.HashMap[String, Object]()
     import kafka.server.KafkaConfig._
-    channelConfigs.put(PrincipalBuilderClassProp, Class.forName(principalBuilderClass))
-    channelConfigs.put(SSLProtocolProp, sslProtocol)
-    channelConfigs.put(SSLEnabledProtocolsProp, sslEnabledProtocols)
-    channelConfigs.put(SSLKeystoreTypeProp, sslKeystoreType)
-    channelConfigs.put(SSLKeystoreLocationProp, sslKeystoreLocation)
-    channelConfigs.put(SSLKeystorePasswordProp, sslKeystorePassword)
-    channelConfigs.put(SSLKeyPasswordProp, sslKeyPassword)
-    channelConfigs.put(SSLTruststoreTypeProp, sslTruststoreType)
-    channelConfigs.put(SSLTruststoreLocationProp, sslTruststoreLocation)
-    channelConfigs.put(SSLTruststorePasswordProp, sslTruststorePassword)
-    channelConfigs.put(SSLKeyManagerAlgorithmProp, sslKeyManagerAlgorithm)
-    channelConfigs.put(SSLTrustManagerAlgorithmProp, sslTrustManagerAlgorithm)
-    channelConfigs.put(SSLClientAuthProp, sslClientAuth)
-    channelConfigs.put(SSLCipherSuitesProp, sslCipher)
+    Seq(
+      (PrincipalBuilderClassProp, Class.forName(principalBuilderClass)),
+      (SSLProtocolProp, sslProtocol),
+      (SSLEnabledProtocolsProp, sslEnabledProtocols),
+      (SSLKeystoreTypeProp, sslKeystoreType),
+      (SSLKeystoreLocationProp, sslKeystoreLocation),
+      (SSLKeystorePasswordProp, sslKeystorePassword),
+      (SSLKeyPasswordProp, sslKeyPassword),
+      (SSLTruststoreTypeProp, sslTruststoreType),
+      (SSLTruststoreLocationProp, sslTruststoreLocation),
+      (SSLTruststorePasswordProp, sslTruststorePassword),
+      (SSLKeyManagerAlgorithmProp, sslKeyManagerAlgorithm),
+      (SSLTrustManagerAlgorithmProp, sslTrustManagerAlgorithm),
+      (SSLClientAuthProp, sslClientAuth),
+      (SSLCipherSuitesProp, sslCipher),
+      (SaslKerberosServiceNameProp, saslKerberosServiceName),
+      (SaslKerberosKinitCmdProp, saslKerberosKinitCmd),
+      (SaslKerberosTicketRenewWindowFactorProp, saslKerberosTicketRenewWindowFactor),
+      (SaslKerberosTicketRenewJitterProp, saslKerberosTicketRenewJitter),
+      (SaslKerberosMinTimeBeforeReloginProp, saslKerberosMinTimeBeforeRelogin),
+      (AuthToLocalProp, authToLocal)
+    ).foreach { case (key, value) =>
+      if (value != null) channelConfigs.put(key, value)
+    }
     channelConfigs
   }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/core/src/main/scala/kafka/server/KafkaServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index f50c266..510957b 100755
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -21,7 +21,7 @@ import java.net.{SocketTimeoutException}
 import java.util
 
 import kafka.admin._
-import kafka.api.{KAFKA_090, ApiVersion}
+import kafka.api.KAFKA_090
 import kafka.log.LogConfig
 import kafka.log.CleanerConfig
 import kafka.log.LogManager
@@ -34,17 +34,16 @@ import kafka.utils._
 import org.apache.kafka.clients.{ManualMetadataUpdater, ClientRequest, NetworkClient}
 import org.apache.kafka.common.Node
 import org.apache.kafka.common.metrics._
-import org.apache.kafka.common.network.{Selectable, ChannelBuilders, NetworkReceive, Selector}
+import org.apache.kafka.common.network.{LoginType, Selectable, ChannelBuilders, NetworkReceive, Selector, Mode}
 import org.apache.kafka.common.protocol.{Errors, ApiKeys, SecurityProtocol}
 import org.apache.kafka.common.metrics.{JmxReporter, Metrics}
 import org.apache.kafka.common.requests.{ControlledShutdownResponse, ControlledShutdownRequest, RequestSend}
 import org.apache.kafka.common.security.JaasUtils
-import org.apache.kafka.common.security.ssl.SSLFactory
 import org.apache.kafka.common.utils.AppInfoParser
 
 import scala.collection.mutable
 import scala.collection.JavaConverters._
-import org.I0Itec.zkclient.{ZkClient, ZkConnection}
+import org.I0Itec.zkclient.ZkClient
 import kafka.controller.{ControllerStats, KafkaController}
 import kafka.cluster.{EndPoint, Broker}
 import kafka.common.{ErrorMapping, InconsistentBrokerIdException, GenerateBrokerIdException}
@@ -308,7 +307,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime, threadNamePr
           "kafka-server-controlled-shutdown",
           Map.empty.asJava,
           false,
-          ChannelBuilders.create(config.interBrokerSecurityProtocol, SSLFactory.Mode.CLIENT, config.channelConfigs)
+          ChannelBuilders.create(config.interBrokerSecurityProtocol, Mode.CLIENT, LoginType.SERVER, config.channelConfigs)
         )
         new NetworkClient(
           selector,
@@ -491,7 +490,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime, threadNamePr
         if (config.interBrokerProtocolVersion.onOrAfter(KAFKA_090))
           networkClientControlledShutdown(config.controlledShutdownMaxRetries.intValue)
         else blockingChannelControlledShutdown(config.controlledShutdownMaxRetries.intValue)
-      
+
       if (!shutdownSucceeded)
         warn("Proceeding to do an unclean shutdown as all the controlled shutdown attempts failed")
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
index 0a17fd0..5aa817d 100644
--- a/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
+++ b/core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
@@ -27,7 +27,7 @@ import kafka.api.KAFKA_090
 import kafka.common.{KafkaStorageException, TopicAndPartition}
 import ReplicaFetcherThread._
 import org.apache.kafka.clients.{ManualMetadataUpdater, NetworkClient, ClientRequest, ClientResponse}
-import org.apache.kafka.common.network.{Selectable, ChannelBuilders, NetworkReceive, Selector}
+import org.apache.kafka.common.network.{LoginType, Selectable, ChannelBuilders, NetworkReceive, Selector, Mode}
 import org.apache.kafka.common.requests.{ListOffsetResponse, FetchResponse, RequestSend, AbstractRequest, ListOffsetRequest}
 import org.apache.kafka.common.requests.{FetchRequest => JFetchRequest}
 import org.apache.kafka.common.{Node, TopicPartition}
@@ -74,7 +74,7 @@ class ReplicaFetcherThread(name: String,
       "replica-fetcher",
       Map("broker-id" -> sourceBroker.id.toString).asJava,
       false,
-      ChannelBuilders.create(brokerConfig.interBrokerSecurityProtocol, SSLFactory.Mode.CLIENT, brokerConfig.channelConfigs)
+      ChannelBuilders.create(brokerConfig.interBrokerSecurityProtocol, Mode.CLIENT, LoginType.SERVER, brokerConfig.channelConfigs)
     )
     new NetworkClient(
       selector,

http://git-wip-us.apache.org/repos/asf/kafka/blob/403158b5/core/src/test/resources/kafka_jaas.conf
----------------------------------------------------------------------
diff --git a/core/src/test/resources/kafka_jaas.conf b/core/src/test/resources/kafka_jaas.conf
new file mode 100644
index 0000000..b097e26
--- /dev/null
+++ b/core/src/test/resources/kafka_jaas.conf
@@ -0,0 +1,29 @@
+/**
+  * 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.
+  */
+KafkaClient {
+	com.sun.security.auth.module.Krb5LoginModule required debug=true
+	useKeyTab=true
+	storeKey=true
+	serviceName="kafka"
+	keyTab="$keytab-location"
+	principal="client@EXAMPLE.COM";
+};
+
+KafkaServer {
+	com.sun.security.auth.module.Krb5LoginModule required debug=true
+	useKeyTab=true
+	storeKey=true
+	serviceName="kafka"
+	keyTab="$keytab-location"
+	principal="kafka/localhost@EXAMPLE.COM";
+};


Mime
View raw message