kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rsiva...@apache.org
Subject kafka git commit: KAFKA-5854; Handle SASL authentication failures as non-retriable exceptions in clients
Date Wed, 20 Sep 2017 21:52:31 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk c2af13ed0 -> 69d2a1771


KAFKA-5854; Handle SASL authentication failures as non-retriable exceptions in clients

This PR implements the client-side of KIP-152, by modifying `KafkaConsumer`, `KafkaProducer`, and `ConsumerGroupCommand` to throw a non-retriable exception when SASL authentication fails.

This PR is co-authored with rajinisivaram.

Author: Vahid Hashemian <vahidhashemian@us.ibm.com>

Reviewers: Jason Gustafson <jason@confluent.io>, Rajini Sivaram <rajinisivaram@googlemail.com>, Ismael Juma <ismael@juma.me.uk>, tedyu <yuzhihong@gmail.com>

Closes #3832 from vahidhashemian/KAFKA-5854


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/69d2a177
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/69d2a177
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/69d2a177

Branch: refs/heads/trunk
Commit: 69d2a177101eb1c29b59b4c64d8c22f6d5e3d281
Parents: c2af13e
Author: Vahid Hashemian <vahidhashemian@us.ibm.com>
Authored: Wed Sep 20 22:51:53 2017 +0100
Committer: Rajini Sivaram <rajinisivaram@googlemail.com>
Committed: Wed Sep 20 22:51:53 2017 +0100

----------------------------------------------------------------------
 .../kafka/clients/ClusterConnectionStates.java  |  44 ++++-
 .../apache/kafka/clients/ConnectionState.java   |   7 +-
 .../org/apache/kafka/clients/KafkaClient.java   |  22 ++-
 .../kafka/clients/ManualMetadataUpdater.java    |  12 ++
 .../java/org/apache/kafka/clients/Metadata.java |  22 ++-
 .../apache/kafka/clients/MetadataUpdater.java   |   8 +
 .../org/apache/kafka/clients/NetworkClient.java |  28 ++-
 .../kafka/clients/consumer/KafkaConsumer.java   |  26 ++-
 .../clients/consumer/OffsetCommitCallback.java  |   2 +-
 .../consumer/internals/AbstractCoordinator.java |   7 +
 .../consumer/internals/ConsumerCoordinator.java |   2 +-
 .../internals/ConsumerNetworkClient.java        |  12 +-
 .../kafka/clients/producer/KafkaProducer.java   |  12 +-
 .../kafka/common/network/ChannelState.java      |  10 +-
 .../org/apache/kafka/clients/MetadataTest.java  |   4 +-
 .../org/apache/kafka/clients/MockClient.java    |   6 +
 .../ClientAuthenticationFailureTest.java        | 114 ++++++++++++
 .../main/scala/kafka/admin/AdminClient.scala    |   7 +-
 .../scala/kafka/tools/ConsoleConsumer.scala     |   5 +-
 .../SaslClientsWithInvalidCredentialsTest.scala | 176 +++++++++++++++++++
 .../SaslScramSslEndToEndAuthorizationTest.scala |  14 +-
 .../scala/integration/kafka/api/SaslSetup.scala |  15 +-
 22 files changed, 496 insertions(+), 59 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java b/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java
index 4d4bedd..5bc25de 100644
--- a/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java
+++ b/clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java
@@ -17,12 +17,15 @@
 package org.apache.kafka.clients;
 
 import java.util.concurrent.ThreadLocalRandom;
+
+import org.apache.kafka.common.errors.AuthenticationException;
+
 import java.util.HashMap;
 import java.util.Map;
 
 /**
  * The state of our connection to each node in the cluster.
- * 
+ *
  */
 final class ClusterConnectionStates {
     private final long reconnectBackoffInitMs;
@@ -50,7 +53,8 @@ final class ClusterConnectionStates {
         if (state == null)
             return true;
         else
-            return state.state == ConnectionState.DISCONNECTED && now - state.lastConnectAttemptMs >= state.reconnectBackoffMs;
+            return state.state.isDisconnected() &&
+                   now - state.lastConnectAttemptMs >= state.reconnectBackoffMs;
     }
 
     /**
@@ -63,7 +67,8 @@ final class ClusterConnectionStates {
         if (state == null)
             return false;
         else
-            return state.state == ConnectionState.DISCONNECTED && now - state.lastConnectAttemptMs < state.reconnectBackoffMs;
+            return state.state.isDisconnected() &&
+                   now - state.lastConnectAttemptMs < state.reconnectBackoffMs;
     }
 
     /**
@@ -77,7 +82,7 @@ final class ClusterConnectionStates {
         NodeConnectionState state = nodeState.get(id);
         if (state == null) return 0;
         long timeWaited = now - state.lastConnectAttemptMs;
-        if (state.state == ConnectionState.DISCONNECTED) {
+        if (state.state.isDisconnected()) {
             return Math.max(state.reconnectBackoffMs - timeWaited, 0);
         } else {
             // When connecting or connected, we should be able to delay indefinitely since other events (connection or
@@ -136,6 +141,20 @@ final class ClusterConnectionStates {
     }
 
     /**
+     * Enter the authentication failed state for the given node.
+     * @param id the connection identifier
+     * @param now the current time
+     * @param exception the authentication exception
+     */
+    public void authenticationFailed(String id, long now, AuthenticationException exception) {
+        NodeConnectionState nodeState = nodeState(id);
+        nodeState.authenticationException = exception;
+        nodeState.state = ConnectionState.AUTHENTICATION_FAILED;
+        nodeState.lastConnectAttemptMs = now;
+        updateReconnectBackoff(nodeState);
+    }
+
+    /**
      * Return true if the connection is ready.
      * @param id the connection identifier
      */
@@ -162,7 +181,16 @@ final class ClusterConnectionStates {
      */
     public boolean isDisconnected(String id) {
         NodeConnectionState state = nodeState.get(id);
-        return state != null && state.state == ConnectionState.DISCONNECTED;
+        return state != null && state.state.isDisconnected();
+    }
+
+    /**
+     * Return authentication exception if an authentication error occurred
+     * @param id The id of the node to check
+     */
+    public AuthenticationException authenticationException(String id) {
+        NodeConnectionState state = nodeState.get(id);
+        return state != null ? state.authenticationException : null;
     }
 
     /**
@@ -205,7 +233,7 @@ final class ClusterConnectionStates {
     public void remove(String id) {
         nodeState.remove(id);
     }
-    
+
     /**
      * Get the state of a given connection.
      * @param id the id of the connection
@@ -225,19 +253,21 @@ final class ClusterConnectionStates {
             throw new IllegalStateException("No entry found for connection " + id);
         return state;
     }
-    
+
     /**
      * The state of our connection to a node.
      */
     private static class NodeConnectionState {
 
         ConnectionState state;
+        AuthenticationException authenticationException;
         long lastConnectAttemptMs;
         long failedAttempts;
         long reconnectBackoffMs;
 
         public NodeConnectionState(ConnectionState state, long lastConnectAttempt, long reconnectBackoffMs) {
             this.state = state;
+            this.authenticationException = null;
             this.lastConnectAttemptMs = lastConnectAttempt;
             this.failedAttempts = 0;
             this.reconnectBackoffMs = reconnectBackoffMs;

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/main/java/org/apache/kafka/clients/ConnectionState.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/ConnectionState.java b/clients/src/main/java/org/apache/kafka/clients/ConnectionState.java
index 62ffada..28b43d6 100644
--- a/clients/src/main/java/org/apache/kafka/clients/ConnectionState.java
+++ b/clients/src/main/java/org/apache/kafka/clients/ConnectionState.java
@@ -23,7 +23,12 @@ package org.apache.kafka.clients;
  * CONNECTING: connection is under progress
  * CHECKING_API_VERSIONS: connection has been established and api versions check is in progress. Failure of this check will cause connection to close
  * READY: connection is ready to send requests
+ * AUTHENTICATION_FAILED: connection failed due to an authentication error
  */
 public enum ConnectionState {
-    DISCONNECTED, CONNECTING, CHECKING_API_VERSIONS, READY
+    DISCONNECTED, CONNECTING, CHECKING_API_VERSIONS, READY, AUTHENTICATION_FAILED;
+
+    public boolean isDisconnected() {
+        return this == AUTHENTICATION_FAILED || this == DISCONNECTED;
+    }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java b/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
index 2faebfd..0a9b519 100644
--- a/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
@@ -17,6 +17,7 @@
 package org.apache.kafka.clients;
 
 import org.apache.kafka.common.Node;
+import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.requests.AbstractRequest;
 
 import java.io.Closeable;
@@ -30,7 +31,7 @@ public interface KafkaClient extends Closeable {
     /**
      * Check if we are currently ready to send another request to the given node but don't attempt to connect if we
      * aren't.
-     * 
+     *
      * @param node The node to check
      * @param now The current timestamp
      */
@@ -39,7 +40,7 @@ public interface KafkaClient extends Closeable {
     /**
      * Initiate a connection to the given node (if necessary), and return true if already connected. The readiness of a
      * node will change only when poll is invoked.
-     * 
+     *
      * @param node The node to connect to.
      * @param now The current time
      * @return true iff we are ready to immediately initiate the sending of another request to the given node.
@@ -50,7 +51,7 @@ public interface KafkaClient extends Closeable {
      * Returns the number of milliseconds to wait, based on the connection state, before attempting to send data. When
      * disconnected, this respects the reconnect backoff time. When connecting or connected, this handles slow/stalled
      * connections.
-     * 
+     *
      * @param node The node to check
      * @param now The current timestamp
      * @return The number of milliseconds to wait.
@@ -68,6 +69,15 @@ public interface KafkaClient extends Closeable {
     boolean connectionFailed(Node node);
 
     /**
+     * Check if authentication to this node has failed, based on the connection state. Authentication failures are
+     * propagated without any retries.
+     *
+     * @param node the node to check
+     * @return an AuthenticationException iff authentication has failed, null otherwise
+     */
+    AuthenticationException authenticationException(Node node);
+
+    /**
      * Queue up the given request for sending. Requests can only be sent on ready connections.
      * @param request The request
      * @param now The current timestamp
@@ -76,7 +86,7 @@ public interface KafkaClient extends Closeable {
 
     /**
      * Do actual reads and writes from sockets.
-     * 
+     *
      * @param timeout The maximum amount of time to wait for responses in ms, must be non-negative. The implementation
      *                is free to use a lower value if appropriate (common reasons for this are a lower request or
      *                metadata update timeout)
@@ -106,7 +116,7 @@ public interface KafkaClient extends Closeable {
      * Choose the node with the fewest outstanding requests. This method will prefer a node with an existing connection,
      * but will potentially choose a node for which we don't yet have a connection if all existing connections are in
      * use.
-     * 
+     *
      * @param now The current time in ms
      * @return The node with the fewest in-flight requests.
      */
@@ -124,7 +134,7 @@ public interface KafkaClient extends Closeable {
 
     /**
      * Get the total in-flight requests for a particular node
-     * 
+     *
      * @param nodeId The id of the node
      */
     int inFlightRequestCount(String nodeId);

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/main/java/org/apache/kafka/clients/ManualMetadataUpdater.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/ManualMetadataUpdater.java b/clients/src/main/java/org/apache/kafka/clients/ManualMetadataUpdater.java
index ed149fb..8252cf3 100644
--- a/clients/src/main/java/org/apache/kafka/clients/ManualMetadataUpdater.java
+++ b/clients/src/main/java/org/apache/kafka/clients/ManualMetadataUpdater.java
@@ -17,8 +17,11 @@
 package org.apache.kafka.clients;
 
 import org.apache.kafka.common.Node;
+import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.requests.MetadataResponse;
 import org.apache.kafka.common.requests.RequestHeader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -34,6 +37,8 @@ import java.util.List;
  */
 public class ManualMetadataUpdater implements MetadataUpdater {
 
+    private static final Logger log = LoggerFactory.getLogger(ManualMetadataUpdater.class);
+
     private List<Node> nodes;
 
     public ManualMetadataUpdater() {
@@ -69,6 +74,13 @@ public class ManualMetadataUpdater implements MetadataUpdater {
     }
 
     @Override
+    public void handleAuthenticationFailure(AuthenticationException exception) {
+        // We don't fail the broker on authentication failures, but there is sufficient information in the broker logs
+        // to identify the failure.
+        log.debug("An authentication error occurred in broker-to-broker communication.", exception);
+    }
+
+    @Override
     public void handleCompletedMetadataResponse(RequestHeader requestHeader, long now, MetadataResponse response) {
         // Do nothing
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/main/java/org/apache/kafka/clients/Metadata.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/Metadata.java b/clients/src/main/java/org/apache/kafka/clients/Metadata.java
index b4521c3..5790d88 100644
--- a/clients/src/main/java/org/apache/kafka/clients/Metadata.java
+++ b/clients/src/main/java/org/apache/kafka/clients/Metadata.java
@@ -20,6 +20,7 @@ import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.internals.ClusterResourceListeners;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.errors.TimeoutException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -59,6 +60,7 @@ public final class Metadata {
     private int version;
     private long lastRefreshMs;
     private long lastSuccessfulRefreshMs;
+    private AuthenticationException authenticationException;
     private Cluster cluster;
     private boolean needUpdate;
     /* Topics with expiry time */
@@ -146,15 +148,28 @@ public final class Metadata {
     }
 
     /**
+     * If any non-retriable authentication exceptions were encountered during
+     * metadata update, clear and throw the exception.
+     */
+    public synchronized void maybeThrowAuthenticationException() {
+        if (authenticationException != null) {
+            AuthenticationException exception = authenticationException;
+            authenticationException = null;
+            throw exception;
+        }
+    }
+
+    /**
      * Wait for metadata update until the current version is larger than the last version we know of
      */
     public synchronized void awaitUpdate(final int lastVersion, final long maxWaitMs) throws InterruptedException {
         if (maxWaitMs < 0) {
-            throw new IllegalArgumentException("Max time to wait for metadata updates should not be < 0 milli seconds");
+            throw new IllegalArgumentException("Max time to wait for metadata updates should not be < 0 milliseconds");
         }
         long begin = System.currentTimeMillis();
         long remainingWaitMs = maxWaitMs;
         while (this.version <= lastVersion) {
+            maybeThrowAuthenticationException();
             if (remainingWaitMs != 0)
                 wait(remainingWaitMs);
             long elapsed = System.currentTimeMillis() - begin;
@@ -256,8 +271,11 @@ public final class Metadata {
      * Record an attempt to update the metadata that failed. We need to keep track of this
      * to avoid retrying immediately.
      */
-    public synchronized void failedUpdate(long now) {
+    public synchronized void failedUpdate(long now, AuthenticationException authenticationException) {
         this.lastRefreshMs = now;
+        this.authenticationException = authenticationException;
+        if (authenticationException != null)
+            this.notifyAll();
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java b/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java
index 55901b5..cb821d6 100644
--- a/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java
+++ b/clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java
@@ -17,6 +17,7 @@
 package org.apache.kafka.clients;
 
 import org.apache.kafka.common.Node;
+import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.requests.MetadataResponse;
 import org.apache.kafka.common.requests.RequestHeader;
 
@@ -62,6 +63,13 @@ interface MetadataUpdater {
     void handleDisconnection(String destination);
 
     /**
+     * Handle authentication failure. Propagate the authentication exception if awaiting metadata.
+     *
+     * @param exception authentication exception from broker
+     */
+    void handleAuthenticationFailure(AuthenticationException exception);
+
+    /**
      * If `request` is a metadata request, handles it and returns `true`. Otherwise, returns `false`.
      *
      * This provides a mechanism for the `MetadataUpdater` implementation to use the NetworkClient instance for its own

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
index c3c15df..f8da42c 100644
--- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -18,6 +18,7 @@ package org.apache.kafka.clients;
 
 import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.Node;
+import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.network.ChannelState;
@@ -313,6 +314,18 @@ public class NetworkClient implements KafkaClient {
     }
 
     /**
+     * Check if authentication to this node has failed, based on the connection state. Authentication failures are
+     * propagated without any retries.
+     *
+     * @param node the node to check
+     * @return an AuthenticationException iff authentication has failed, null otherwise
+     */
+    @Override
+    public AuthenticationException authenticationException(Node node) {
+        return connectionStates.authenticationException(node.idString());
+    }
+
+    /**
      * Check if the node with the given id is ready to send more requests.
      *
      * @param node The node
@@ -589,6 +602,7 @@ public class NetworkClient implements KafkaClient {
         nodesNeedingApiVersionsFetch.remove(nodeId);
         switch (disconnectState.state()) {
             case AUTHENTICATION_FAILED:
+                connectionStates.authenticationFailed(nodeId, now, disconnectState.exception());
                 log.error("Connection to node {} failed authentication due to: {}", nodeId, disconnectState.exception().getMessage());
                 break;
             case AUTHENTICATE:
@@ -610,6 +624,9 @@ public class NetworkClient implements KafkaClient {
             else
                 responses.add(request.disconnected(now));
         }
+        AuthenticationException authenticationException = connectionStates.authenticationException(nodeId);
+        if (authenticationException != null)
+            metadataUpdater.handleAuthenticationFailure(authenticationException);
     }
 
     /**
@@ -848,6 +865,13 @@ public class NetworkClient implements KafkaClient {
         }
 
         @Override
+        public void handleAuthenticationFailure(AuthenticationException exception) {
+            metadataFetchInProgress = false;
+            if (metadata.updateRequested())
+                metadata.failedUpdate(time.milliseconds(), exception);
+        }
+
+        @Override
         public void handleCompletedMetadataResponse(RequestHeader requestHeader, long now, MetadataResponse response) {
             this.metadataFetchInProgress = false;
             Cluster cluster = response.cluster();
@@ -862,7 +886,7 @@ public class NetworkClient implements KafkaClient {
                 this.metadata.update(cluster, response.unavailableTopics(), now);
             } else {
                 log.trace("Ignoring empty metadata response with correlation id {}.", requestHeader.correlationId());
-                this.metadata.failedUpdate(now);
+                this.metadata.failedUpdate(now, null);
             }
         }
 
@@ -979,7 +1003,7 @@ public class NetworkClient implements KafkaClient {
         public ClientResponse disconnected(long timeMs) {
             return new ClientResponse(header, callback, destination, createdTimeMs, timeMs, true, null, null);
         }
-        
+
         @Override
         public String toString() {
             return "InFlightRequest(header=" + header +

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
index c2f2f5f..3ea0394 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
@@ -1064,8 +1064,9 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
      *             function is called
      * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
      *             this function is called
+     * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
      * @throws org.apache.kafka.common.errors.AuthorizationException if caller lacks Read access to any of the subscribed
-     *             topics or to the configured groupId
+     *             topics or to the configured groupId. See the exception for more details
      * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors (e.g. invalid groupId or
      *             session timeout, errors deserializing key/value pairs, or any new error cases in future versions)
      * @throws java.lang.IllegalArgumentException if the timeout value is negative
@@ -1176,8 +1177,9 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
      *             function is called
      * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
      *             this function is called
+     * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
      * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the
-     *             configured groupId
+     *             configured groupId. See the exception for more details
      * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors (e.g. if offset metadata
      *             is too large or if the topic does not exist).
      */
@@ -1213,8 +1215,9 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
      *             function is called
      * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
      *             this function is called
+     * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
      * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the
-     *             configured groupId
+     *             configured groupId. See the exception for more details
      * @throws java.lang.IllegalArgumentException if the committed offset is negative
      * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors (e.g. if offset metadata
      *             is too large or if the topic does not exist).
@@ -1379,8 +1382,9 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
      *             function is called
      * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
      *             this function is called
+     * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
      * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the
-     *             configured groupId
+     *             configured groupId. See the exception for more details
      * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors
      */
     public long position(TopicPartition partition) {
@@ -1412,8 +1416,9 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
      *             function is called
      * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
      *             this function is called
+     * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
      * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the
-     *             configured groupId
+     *             configured groupId. See the exception for more details
      * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors
      */
     @Override
@@ -1445,7 +1450,8 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
      *             function is called
      * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
      *             this function is called
-     * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the specified topic
+     * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
+     * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the specified topic. See the exception for more details
      * @throws org.apache.kafka.common.errors.TimeoutException if the topic metadata could not be fetched before
      *             expiration of the configured request timeout
      * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors
@@ -1560,6 +1566,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
      * @return a mapping from partition to the timestamp and offset of the first message with timestamp greater
      *         than or equal to the target timestamp. {@code null} will be returned for the partition if there is no
      *         such message.
+     * @throws AuthenticationException if authentication fails. See the exception for more details
      * @throws IllegalArgumentException if the target timestamp is negative.
      * @throws org.apache.kafka.common.errors.TimeoutException if the offset metadata could not be fetched before
      *         expiration of the configured request timeout
@@ -1593,6 +1600,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
      *
      * @param partitions the partitions to get the earliest offsets.
      * @return The earliest available offsets for the given partitions
+     * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
      * @throws org.apache.kafka.common.errors.TimeoutException if the offset metadata could not be fetched before
      *         expiration of the configured request timeout
      */
@@ -1621,6 +1629,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
      *
      * @param partitions the partitions to get the end offsets.
      * @return The end offsets for the given partitions.
+     * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
      * @throws org.apache.kafka.common.errors.TimeoutException if the offset metadata could not be fetched before
      *         expiration of the configured request timeout
      */
@@ -1641,8 +1650,9 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
      * timeout. See {@link #close(long, TimeUnit)} for details. Note that {@link #wakeup()}
      * cannot be used to interrupt close.
      *
+     * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
      * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted
-     * before or while this function is called
+     *             before or while this function is called
      */
     @Override
     public void close() {
@@ -1660,6 +1670,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
      * @param timeout The maximum time to wait for consumer to close gracefully. The value must be
      *                non-negative. Specifying a timeout of zero means do not wait for pending requests to complete.
      * @param timeUnit The time unit for the {@code timeout}
+     * @throws AuthenticationException if authentication fails. See the exception for more details
      * @throws InterruptException If the thread is interrupted before or while this function is called
      * @throws IllegalArgumentException If the {@code timeout} is negative.
      */
@@ -1729,6 +1740,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
      * or reset it using the offset reset policy the user has configured.
      *
      * @param partitions The partitions that needs updating fetch positions
+     * @throws AuthenticationException if authentication fails. See the exception for more details
      * @throws NoOffsetForPartitionException If no offset is stored for a given partition and no offset reset policy is
      *             defined
      */

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetCommitCallback.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetCommitCallback.java b/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetCommitCallback.java
index 2fef79e..b217a63 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetCommitCallback.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/OffsetCommitCallback.java
@@ -42,7 +42,7 @@ public interface OffsetCommitCallback {
      * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
      *             this function is called
      * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the
-     *             configured groupId
+     *             configured groupId. See the exception for more details
      * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors (e.g. if offset metadata
      *             is too large or if the committed offset is invalid).
      */

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
index bf75242..86e4f2b 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
@@ -19,6 +19,7 @@ package org.apache.kafka.clients.consumer.internals;
 import org.apache.kafka.clients.ClientResponse;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.Node;
+import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.errors.DisconnectException;
 import org.apache.kafka.common.errors.GroupAuthorizationException;
 import org.apache.kafka.common.errors.IllegalGenerationException;
@@ -982,6 +983,12 @@ public abstract class AbstractCoordinator implements Closeable {
                         }
                     }
                 }
+            } catch (AuthenticationException e) {
+                log.error("An authentication error occurred in the heartbeat thread", e);
+                this.failed.set(e);
+            } catch (GroupAuthorizationException e) {
+                log.error("A group authorization error occurred in the heartbeat thread for group {}", groupId, e);
+                this.failed.set(e);
             } catch (InterruptedException | InterruptException e) {
                 Thread.interrupted();
                 log.error("Unexpected interrupt received in heartbeat thread for group {}", groupId, e);

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
index e740ba7..38ca041 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
@@ -574,7 +574,7 @@ public final class ConsumerCoordinator extends AbstractCoordinator {
      * or an unrecoverable error is encountered.
      * @param offsets The offsets to be committed
      * @throws org.apache.kafka.common.errors.AuthorizationException if the consumer is not authorized to the group
-     *             or to any of the specified partitions
+     *             or to any of the specified partitions. See the exception for more details
      * @throws CommitFailedException if an unrecoverable error occurs before the commit can be completed
      * @return If the offset commit was successfully sent and a successful response was received from
      *         the coordinator

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
index 803a853..43c6358 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
@@ -22,6 +22,7 @@ import org.apache.kafka.clients.KafkaClient;
 import org.apache.kafka.clients.Metadata;
 import org.apache.kafka.clients.RequestCompletionHandler;
 import org.apache.kafka.common.Node;
+import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.errors.DisconnectException;
 import org.apache.kafka.common.errors.InterruptException;
 import org.apache.kafka.common.errors.TimeoutException;
@@ -134,6 +135,7 @@ public class ConsumerNetworkClient implements Closeable {
         int version = this.metadata.requestUpdate();
         do {
             poll(timeout);
+            this.metadata.maybeThrowAuthenticationException();
         } while (this.metadata.version() == version && time.milliseconds() - startMs < timeout);
         return this.metadata.version() > version;
     }
@@ -367,9 +369,13 @@ public class ConsumerNetworkClient implements Closeable {
                 Collection<ClientRequest> requests = unsent.remove(node);
                 for (ClientRequest request : requests) {
                     RequestFutureCompletionHandler handler = (RequestFutureCompletionHandler) request.callback();
-                    handler.onComplete(new ClientResponse(request.makeHeader(request.requestBuilder().desiredOrLatestVersion()),
-                        request.callback(), request.destination(), request.createdTimeMs(), now, true,
-                        null, null));
+                    AuthenticationException authenticationException = client.authenticationException(node);
+                    if (authenticationException != null)
+                        handler.onFailure(authenticationException);
+                    else
+                        handler.onComplete(new ClientResponse(request.makeHeader(request.requestBuilder().desiredOrLatestVersion()),
+                            request.callback(), request.destination(), request.createdTimeMs(), now, true,
+                            null, null));
                 }
             }
         }

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
index 7dcec5c..66760e2 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
@@ -537,7 +537,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
      * @throws org.apache.kafka.common.errors.UnsupportedVersionException fatal error indicating the broker
      *         does not support transactions (i.e. if its version is lower than 0.11.0.0)
      * @throws org.apache.kafka.common.errors.AuthorizationException fatal error indicating that the configured
-     *         transactional.id is not authorized
+     *         transactional.id is not authorized. See the exception for more details
      * @throws KafkaException if the producer has encountered a previous fatal error or for any other unexpected error
      */
     public void initTransactions() {
@@ -557,7 +557,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
      * @throws org.apache.kafka.common.errors.UnsupportedVersionException fatal error indicating the broker
      *         does not support transactions (i.e. if its version is lower than 0.11.0.0)
      * @throws org.apache.kafka.common.errors.AuthorizationException fatal error indicating that the configured
-     *         transactional.id is not authorized
+     *         transactional.id is not authorized. See the exception for more details
      * @throws KafkaException if the producer has encountered a previous fatal error or for any other unexpected error
      */
     public void beginTransaction() throws ProducerFencedException {
@@ -585,7 +585,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
      * @throws org.apache.kafka.common.errors.UnsupportedForMessageFormatException  fatal error indicating the message
      *         format used for the offsets topic on the broker does not support transactions
      * @throws org.apache.kafka.common.errors.AuthorizationException fatal error indicating that the configured
-     *         transactional.id is not authorized
+     *         transactional.id is not authorized. See the exception for more details
      * @throws KafkaException if the producer has encountered a previous fatal or abortable error, or for any
      *         other unexpected error
      */
@@ -609,7 +609,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
      * @throws org.apache.kafka.common.errors.UnsupportedVersionException fatal error indicating the broker
      *         does not support transactions (i.e. if its version is lower than 0.11.0.0)
      * @throws org.apache.kafka.common.errors.AuthorizationException fatal error indicating that the configured
-     *         transactional.id is not authorized
+     *         transactional.id is not authorized. See the exception for more details
      * @throws KafkaException if the producer has encountered a previous fatal or abortable error, or for any
      *         other unexpected error
      */
@@ -630,7 +630,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
      * @throws org.apache.kafka.common.errors.UnsupportedVersionException fatal error indicating the broker
      *         does not support transactions (i.e. if its version is lower than 0.11.0.0)
      * @throws org.apache.kafka.common.errors.AuthorizationException fatal error indicating that the configured
-     *         transactional.id is not authorized
+     *         transactional.id is not authorized. See the exception for more details
      * @throws KafkaException if the producer has encountered a previous fatal error or for any other unexpected error
      */
     public void abortTransaction() throws ProducerFencedException {
@@ -746,6 +746,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
      * @param callback A user-supplied callback to execute when the record has been acknowledged by the server (null
      *        indicates no callback)
      *
+     * @throws AuthenticationException if authentication fails. See the exception for more details
      * @throws IllegalStateException if a transactional.id has been configured and no transaction has been started
      * @throws InterruptException If the thread is interrupted while blocked
      * @throws SerializationException If the key or value are not valid objects given the configured serializers
@@ -968,6 +969,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
 
     /**
      * Get the partition metadata for the given topic. This can be used for custom partitioning.
+     * @throws AuthenticationException if authentication fails. See the exception for more details
      * @throws InterruptException If the thread is interrupted while blocked
      */
     @Override

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/main/java/org/apache/kafka/common/network/ChannelState.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/ChannelState.java b/clients/src/main/java/org/apache/kafka/common/network/ChannelState.java
index 4370fd8..08ed1a0 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/ChannelState.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/ChannelState.java
@@ -16,7 +16,7 @@
  */
 package org.apache.kafka.common.network;
 
-import org.apache.kafka.common.errors.ApiException;
+import org.apache.kafka.common.errors.AuthenticationException;
 
 /**
  * States for KafkaChannel:
@@ -73,11 +73,11 @@ public class ChannelState {
     public static final ChannelState LOCAL_CLOSE = new ChannelState(State.LOCAL_CLOSE);
 
     private final State state;
-    private final ApiException exception;
+    private final AuthenticationException exception;
     public ChannelState(State state) {
         this(state, null);
     }
-    public ChannelState(State state, ApiException exception) {
+    public ChannelState(State state, AuthenticationException exception) {
         this.state = state;
         this.exception = exception;
     }
@@ -86,7 +86,7 @@ public class ChannelState {
         return state;
     }
 
-    public ApiException exception() {
+    public AuthenticationException exception() {
         return exception;
     }
-}
\ No newline at end of file
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/test/java/org/apache/kafka/clients/MetadataTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/MetadataTest.java b/clients/src/test/java/org/apache/kafka/clients/MetadataTest.java
index 407eb9f..3f2a94c 100644
--- a/clients/src/test/java/org/apache/kafka/clients/MetadataTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/MetadataTest.java
@@ -135,7 +135,7 @@ public class MetadataTest {
         long now = 10000;
 
         // lastRefreshMs updated to now.
-        metadata.failedUpdate(now);
+        metadata.failedUpdate(now, null);
 
         // Backing off. Remaining time until next try should be returned.
         assertEquals(refreshBackoffMs, metadata.timeToNextUpdate(now));
@@ -216,7 +216,7 @@ public class MetadataTest {
         metadata.update(Cluster.empty(), Collections.<String>emptySet(), time);
 
         assertEquals(100, metadata.timeToNextUpdate(1000));
-        metadata.failedUpdate(1100);
+        metadata.failedUpdate(1100, null);
 
         assertEquals(100, metadata.timeToNextUpdate(1100));
         assertEquals(100, metadata.lastSuccessfulUpdate());

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/test/java/org/apache/kafka/clients/MockClient.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/MockClient.java b/clients/src/test/java/org/apache/kafka/clients/MockClient.java
index 71e32ff..66ff253 100644
--- a/clients/src/test/java/org/apache/kafka/clients/MockClient.java
+++ b/clients/src/test/java/org/apache/kafka/clients/MockClient.java
@@ -18,6 +18,7 @@ package org.apache.kafka.clients;
 
 import org.apache.kafka.common.Cluster;
 import org.apache.kafka.common.Node;
+import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.requests.AbstractRequest;
 import org.apache.kafka.common.requests.AbstractResponse;
@@ -135,6 +136,11 @@ public class MockClient implements KafkaClient {
     }
 
     @Override
+    public AuthenticationException authenticationException(Node node) {
+        return null;
+    }
+
+    @Override
     public void disconnect(String node) {
         long now = time.milliseconds();
         Iterator<ClientRequest> iter = requests.iterator();

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/clients/src/test/java/org/apache/kafka/common/security/authenticator/ClientAuthenticationFailureTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/ClientAuthenticationFailureTest.java b/clients/src/test/java/org/apache/kafka/common/security/authenticator/ClientAuthenticationFailureTest.java
new file mode 100644
index 0000000..b4460f7
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/ClientAuthenticationFailureTest.java
@@ -0,0 +1,114 @@
+/*
+ * 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.authenticator;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.config.SaslConfigs;
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
+import org.apache.kafka.common.errors.AuthenticationFailedException;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.network.NetworkTestUtils;
+import org.apache.kafka.common.network.NioEchoServer;
+import org.apache.kafka.common.protocol.SecurityProtocol;
+import org.apache.kafka.common.security.TestSecurityConfig;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class ClientAuthenticationFailureTest {
+
+    private NioEchoServer server;
+    private Map<String, Object> saslServerConfigs;
+    private Map<String, Object> saslClientConfigs;
+    private final String topic = "test";
+    private TestJaasConfig testJaasConfig;
+
+    @Before
+    public void setup() throws Exception {
+        SecurityProtocol securityProtocol = SecurityProtocol.SASL_PLAINTEXT;
+
+        saslServerConfigs = new HashMap<>();
+        saslServerConfigs.put(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, Arrays.asList("PLAIN"));
+
+        saslClientConfigs = new HashMap<>();
+        saslClientConfigs.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
+        saslClientConfigs.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
+
+        testJaasConfig = TestJaasConfig.createConfiguration("PLAIN", Arrays.asList("PLAIN"));
+        testJaasConfig.setClientOptions("PLAIN", TestJaasConfig.USERNAME, "anotherpassword");
+        server = createEchoServer(securityProtocol);
+    }
+
+    @After
+    public void teardown() throws Exception {
+        if (server != null)
+            server.close();
+    }
+
+    @Test
+    public void testConsumerWithInvalidCredentials() {
+        saslClientConfigs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:" + server.port());
+        saslClientConfigs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
+        saslClientConfigs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
+
+        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(saslClientConfigs)) {
+            consumer.subscribe(Arrays.asList(topic));
+            consumer.poll(100);
+            fail("Expected an authentication error!");
+        } catch (AuthenticationFailedException e) {
+            // OK
+        } catch (Exception e) {
+            fail("Expected only an authentication error, but another error occurred: " + e.getMessage());
+        }
+    }
+
+    @Test
+    public void testProducerWithInvalidCredentials() {
+        saslClientConfigs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:" + server.port());
+        saslClientConfigs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+        saslClientConfigs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+
+        ProducerRecord<String, String> record = new ProducerRecord<>(topic, "message");
+        try (KafkaProducer<String, String> producer = new KafkaProducer<>(saslClientConfigs)) {
+            producer.send(record).get();
+            fail("Expected an authentication error!");
+        } catch (Exception e) {
+            assertTrue("Expected an exception of type AuthenticationFailedException", e.getCause() instanceof AuthenticationFailedException);
+        }
+    }
+
+    private NioEchoServer createEchoServer(SecurityProtocol securityProtocol) throws Exception {
+        return createEchoServer(ListenerName.forSecurityProtocol(securityProtocol), securityProtocol);
+    }
+
+    private NioEchoServer createEchoServer(ListenerName listenerName, SecurityProtocol securityProtocol) throws Exception {
+        return NetworkTestUtils.createEchoServer(listenerName, securityProtocol,
+                new TestSecurityConfig(saslServerConfigs), new CredentialCache());
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/core/src/main/scala/kafka/admin/AdminClient.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/admin/AdminClient.scala b/core/src/main/scala/kafka/admin/AdminClient.scala
index fadacec..24149d7 100644
--- a/core/src/main/scala/kafka/admin/AdminClient.scala
+++ b/core/src/main/scala/kafka/admin/AdminClient.scala
@@ -26,7 +26,7 @@ import org.apache.kafka.clients._
 import org.apache.kafka.clients.consumer.internals.{ConsumerNetworkClient, ConsumerProtocol, RequestFuture, RequestFutureAdapter}
 import org.apache.kafka.common.config.ConfigDef.{Importance, Type}
 import org.apache.kafka.common.config.{AbstractConfig, ConfigDef}
-import org.apache.kafka.common.errors.TimeoutException
+import org.apache.kafka.common.errors.{AuthenticationException, TimeoutException}
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.network.Selector
 import org.apache.kafka.common.protocol.{ApiKeys, Errors}
@@ -57,9 +57,8 @@ class AdminClient(val time: Time,
   val networkThread = new KafkaThread("admin-client-network-thread", new Runnable {
     override def run() {
       try {
-        while (running) {
+        while (running)
           client.poll(Long.MaxValue)
-        }
       } catch {
         case t : Throwable =>
           error("admin-client-network-thread exited", t)
@@ -96,6 +95,8 @@ class AdminClient(val time: Time,
       try {
         return send(broker, api, request)
       } catch {
+        case e: AuthenticationException =>
+          throw e
         case e: Exception =>
           debug(s"Request $api failed against node $broker", e)
       }

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/core/src/main/scala/kafka/tools/ConsoleConsumer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/tools/ConsoleConsumer.scala b/core/src/main/scala/kafka/tools/ConsoleConsumer.scala
index a0818bc..c014caf 100755
--- a/core/src/main/scala/kafka/tools/ConsoleConsumer.scala
+++ b/core/src/main/scala/kafka/tools/ConsoleConsumer.scala
@@ -31,7 +31,7 @@ import kafka.metrics.KafkaMetricsReporter
 import kafka.utils._
 import kafka.utils.Implicits._
 import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
-import org.apache.kafka.common.errors.WakeupException
+import org.apache.kafka.common.errors.{AuthenticationException, WakeupException}
 import org.apache.kafka.common.record.TimestampType
 import org.apache.kafka.common.serialization.Deserializer
 import org.apache.kafka.common.utils.Utils
@@ -53,6 +53,9 @@ object ConsoleConsumer extends Logging {
     try {
       run(conf)
     } catch {
+      case e: AuthenticationException =>
+        error("Authentication failed: terminating consumer process", e)
+        Exit.exit(1)
       case e: Throwable =>
         error("Unknown error when running consumer: ", e)
         Exit.exit(1)

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala b/core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala
new file mode 100644
index 0000000..52fbdba
--- /dev/null
+++ b/core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala
@@ -0,0 +1,176 @@
+/**
+  * 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 kafka.api
+
+import java.io.FileOutputStream
+import java.util.concurrent.{ExecutionException, Future, TimeUnit}
+import scala.collection.JavaConverters.seqAsJavaListConverter
+
+import org.apache.kafka.clients.consumer.{KafkaConsumer, ConsumerConfig}
+import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, RecordMetadata}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.AuthenticationFailedException
+import org.apache.kafka.common.protocol.SecurityProtocol
+import org.apache.kafka.common.serialization.ByteArrayDeserializer
+import org.junit.{After, Before, Test}
+import org.junit.Assert._
+
+import kafka.admin.ConsumerGroupCommand.{ConsumerGroupCommandOptions, KafkaConsumerGroupService}
+import kafka.server.KafkaConfig
+import kafka.utils.{JaasTestUtils, TestUtils, ZkUtils}
+
+class SaslClientsWithInvalidCredentialsTest extends IntegrationTestHarness with SaslSetup {
+  private val kafkaClientSaslMechanism = "SCRAM-SHA-256"
+  private val kafkaServerSaslMechanisms = List(kafkaClientSaslMechanism)
+  override protected val securityProtocol = SecurityProtocol.SASL_PLAINTEXT
+  override protected val serverSaslProperties = Some(kafkaServerSaslProperties(kafkaServerSaslMechanisms, kafkaClientSaslMechanism))
+  override protected val clientSaslProperties = Some(kafkaClientSaslProperties(kafkaClientSaslMechanism))
+  val consumerCount = 1
+  val producerCount = 1
+  val serverCount = 1
+
+  this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+
+  val topic = "topic"
+  val tp = new TopicPartition(topic, 0)
+
+  override def configureSecurityBeforeServersStart() {
+    super.configureSecurityBeforeServersStart()
+    zkUtils.makeSurePersistentPathExists(ZkUtils.ConfigChangesPath)
+    // Create broker credentials before starting brokers
+    createScramCredentials(zkConnect, JaasTestUtils.KafkaScramAdmin, JaasTestUtils.KafkaScramAdminPassword)
+  }
+
+  @Before
+  override def setUp(): Unit = {
+    startSasl(jaasSections(kafkaServerSaslMechanisms, Some(kafkaClientSaslMechanism), Both,
+      JaasTestUtils.KafkaServerContextName))
+    super.setUp()
+    TestUtils.createTopic(this.zkUtils, topic, 1, serverCount, this.servers)
+  }
+
+  @After
+  override def tearDown(): Unit = {
+    super.tearDown()
+    closeSasl()
+  }
+
+  @Test
+  def testProducerWithAuthenticationFailure() {
+    verifyAuthenticationException(() => sendOneRecord(10000))
+
+    createClientCredential()
+    verifyWithRetry(() => sendOneRecord())
+  }
+
+  @Test
+  def testConsumerWithAuthenticationFailure() {
+    val consumer = this.consumers.head
+    consumer.subscribe(List(topic).asJava)
+    verifyConsumerWithAuthenticationFailure(consumer)
+  }
+
+  @Test
+  def testManualAssignmentConsumerWithAuthenticationFailure() {
+    val consumer = this.consumers.head
+    consumer.assign(List(tp).asJava)
+    verifyConsumerWithAuthenticationFailure(consumer)
+  }
+
+  @Test
+  def testManualAssignmentConsumerWithAutoCommitDisabledWithAuthenticationFailure() {
+    this.consumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false.toString)
+    val consumer = new KafkaConsumer(this.consumerConfig, new ByteArrayDeserializer(), new ByteArrayDeserializer())
+    consumers += consumer
+    consumer.assign(List(tp).asJava)
+    consumer.seek(tp, 0)
+
+    verifyConsumerWithAuthenticationFailure(consumer)
+  }
+
+  private def verifyConsumerWithAuthenticationFailure(consumer: KafkaConsumer[Array[Byte], Array[Byte]]) {
+    verifyAuthenticationException(() => consumer.poll(10000))
+
+    createClientCredential()
+    verifyWithRetry(() => sendOneRecord())
+    verifyWithRetry(() => assertEquals(1, consumer.poll(1000).count))
+  }
+
+  @Test
+  def testConsumerGroupServiceWithAuthenticationFailure() {
+    val propsFile = TestUtils.tempFile()
+    val propsStream = new FileOutputStream(propsFile)
+    propsStream.write("security.protocol=SASL_PLAINTEXT\n".getBytes())
+    propsStream.write(s"sasl.mechanism=$kafkaClientSaslMechanism".getBytes())
+    propsStream.close()
+
+    val cgcArgs = Array("--bootstrap-server", brokerList,
+                        "--describe",
+                        "--group", "test.group",
+                        "--command-config", propsFile.getAbsolutePath)
+    val opts = new ConsumerGroupCommandOptions(cgcArgs)
+    val consumerGroupService = new KafkaConsumerGroupService(opts)
+
+    val consumer = consumers.head
+    consumer.subscribe(List(topic).asJava)
+
+    verifyAuthenticationException(() => consumerGroupService.listGroups)
+    createClientCredential()
+    verifyWithRetry(() => consumer.poll(1000))
+    assertEquals(1, consumerGroupService.listGroups.size)
+  }
+
+  private def createClientCredential(): Unit = {
+    createScramCredentials(zkConnect, JaasTestUtils.KafkaScramUser2, JaasTestUtils.KafkaScramPassword2)
+  }
+
+  private def sendOneRecord(maxWaitMs: Long = 15000): Unit = {
+    val producer = this.producers.head
+    val record = new ProducerRecord(tp.topic(), tp.partition(), 0L, "key".getBytes, "value".getBytes)
+    val future = producer.send(record)
+    producer.flush()
+    try {
+      val recordMetadata = future.get(maxWaitMs, TimeUnit.MILLISECONDS)
+      assertTrue(s"Invalid offset $recordMetadata", recordMetadata.offset >= 0)
+    } catch {
+      case e: ExecutionException => throw e.getCause
+    }
+  }
+
+  private def verifyAuthenticationException(f: () => Unit): Unit = {
+    val startMs = System.currentTimeMillis
+    try {
+      f()
+      fail("Expected an authentication exception")
+    } catch {
+      case e: AuthenticationFailedException =>
+        // expected exception
+        val elapsedMs = System.currentTimeMillis - startMs
+        assertTrue(s"Poll took too long, elapsed=$elapsedMs", elapsedMs <= 5000)
+        assertTrue(s"Exception message not useful: $e", e.getMessage.contains("invalid credentials"))
+    }
+  }
+
+  private def verifyWithRetry(f: () => Unit): Unit = {
+    var attempts = 0
+    TestUtils.waitUntilTrue(() => {
+      try {
+        attempts += 1
+        f()
+        true
+      } catch {
+        case _: AuthenticationFailedException => false
+      }
+    }, s"Operation did not succeed within timeout after $attempts")
+  }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/core/src/test/scala/integration/kafka/api/SaslScramSslEndToEndAuthorizationTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/SaslScramSslEndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/SaslScramSslEndToEndAuthorizationTest.scala
index 0bc4e50..ed1c77b 100644
--- a/core/src/test/scala/integration/kafka/api/SaslScramSslEndToEndAuthorizationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/SaslScramSslEndToEndAuthorizationTest.scala
@@ -35,22 +35,14 @@ class SaslScramSslEndToEndAuthorizationTest extends SaslEndToEndAuthorizationTes
     super.configureSecurityBeforeServersStart()
     zkUtils.makeSurePersistentPathExists(ZkUtils.ConfigChangesPath)
     // Create broker credentials before starting brokers
-    ConfigCommand.main(configCommandArgs(kafkaPrincipal, kafkaPassword))
+    createScramCredentials(zkConnect, kafkaPrincipal, kafkaPassword)
   }
 
   @Before
   override def setUp() {
     super.setUp()
     // Create client credentials after starting brokers so that dynamic credential creation is also tested
-    ConfigCommand.main(configCommandArgs(clientPrincipal, clientPassword))
-    ConfigCommand.main(configCommandArgs(JaasTestUtils.KafkaScramUser2, JaasTestUtils.KafkaScramPassword2))
-  }
-
-  private def configCommandArgs(username: String, password: String) : Array[String] = {
-    val credentials = kafkaServerSaslMechanisms.map(m => s"$m=[iterations=4096,password=$password]")
-    Array("--zookeeper", zkConnect,
-          "--alter", "--add-config", credentials.mkString(","),
-          "--entity-type", "users",
-          "--entity-name", username)
+    createScramCredentials(zkConnect, clientPrincipal, clientPassword)
+    createScramCredentials(zkConnect, JaasTestUtils.KafkaScramUser2, JaasTestUtils.KafkaScramPassword2)
   }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/69d2a177/core/src/test/scala/integration/kafka/api/SaslSetup.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/SaslSetup.scala b/core/src/test/scala/integration/kafka/api/SaslSetup.scala
index f874f4e..51deb85 100644
--- a/core/src/test/scala/integration/kafka/api/SaslSetup.scala
+++ b/core/src/test/scala/integration/kafka/api/SaslSetup.scala
@@ -21,14 +21,16 @@ import java.io.File
 import java.util.Properties
 import javax.security.auth.login.Configuration
 
+import kafka.admin.ConfigCommand
 import kafka.security.minikdc.MiniKdc
 import kafka.server.KafkaConfig
 import kafka.utils.JaasTestUtils.{JaasSection, Krb5LoginModule, ZkDigestModule}
 import kafka.utils.{JaasTestUtils, TestUtils}
-import org.apache.kafka.common.security.JaasUtils
-import org.apache.kafka.common.security.authenticator.LoginManager
 import org.apache.kafka.common.config.SaslConfigs
 import org.apache.kafka.common.config.internals.BrokerSecurityConfigs
+import org.apache.kafka.common.security.JaasUtils
+import org.apache.kafka.common.security.authenticator.LoginManager
+import org.apache.kafka.common.security.scram.ScramMechanism
 
 /*
  * Implements an enumeration for the modes enabled here:
@@ -135,4 +137,13 @@ trait SaslSetup {
   def jaasClientLoginModule(clientSaslMechanism: String): String =
     JaasTestUtils.clientLoginModule(clientSaslMechanism, clientKeytabFile)
 
+  def createScramCredentials(zkConnect: String, userName: String, password: String): Unit = {
+    val credentials = ScramMechanism.values.map(m => s"${m.mechanismName}=[iterations=4096,password=$password]")
+    val args = Array("--zookeeper", zkConnect,
+      "--alter", "--add-config", credentials.mkString(","),
+      "--entity-type", "users",
+      "--entity-name", userName)
+    ConfigCommand.main(args)
+  }
+
 }


Mime
View raw message