KAFKA-5947: Handle authentication failure in admin client, txn producer #3928

rajinisivaram · 2017-09-20T23:17:51Z

Raise AuthenticationException for authentication failures in admin client
Handle AuthenticationException as a fatal error for transactional producer
Add comments to authentication exceptions

producer

rajinisivaram · 2017-09-20T23:18:31Z

@vahidhashemian @hachikuji Can you review, please? Thanks.

ijuma · 2017-09-21T00:17:45Z

clients/src/main/java/org/apache/kafka/common/errors/AuthenticationFailedException.java

+ * </p><p>
+ * SASL authentication failures typically indicate invalid credentials.
+ * <p>
+ */
 public class AuthenticationFailedException extends AuthenticationException {


I wonder if this should be called InvalidCredentialsException.

At the moment, we have a single error code and exception for authentication failure with an error message giving more details. For SSL, this includes any handshake exception (e.g. server host name verification failure). At the moment, we are reporting only invalid credentials for SASL, but we may add others too. Since we convert SSLException and SaslException from the providers, it is simpler to maintain a more generic AuthenticationFailed exception rather than separate exceptions for different failures.

The issue I have is that AuthenticationFailedException is too similar to AuthenticationException.

What about SaslAuthenticationFailedException and SslAuthenticationFailedException?

That sounds good to me.

I will add without the Failed to be consistent with TopicAuthorizationException etc. So SaslAuthenticationException and SslAuthenticationException.

I read it a bit too quickly on my phone and I hadn't seen the "Failed" part. What you did in the end is what I had in mind.

I removed the SSL exception part from the doc. Will add SslAuthenticationException and the associated docs in #3918

vahidhashemian · 2017-09-21T03:41:41Z

@rajinisivaram Thanks for the PR. It looks good to me after a quick review.
I like the alternate exception name @ijuma suggested.

ijuma

Thanks for the PR,. Looks good overall, just a few minor comments.

ijuma · 2017-09-21T07:50:26Z

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -974,6 +1011,9 @@ public void run() {

                // Update the current time and handle the latest responses.
                now = time.milliseconds();
+                if (handleAuthenticationException(now, callsToSend) &&
+                        hardShutdownTimeMs.get() != INVALID_SHUTDOWN_TIME)


What is the reason for hardShutdownTimeMs.get() != INVALID_SHUTDOWN_TIME?

We don't want to exit this background thread until a close has been requested. If we are closing and there is an authentication exception, we want to terminate immediately. If we are not closing, the thread is kept alive so that a future request after credentials are added can succeed. This is tested in the integration test.

Hmm, what's the impact of not exiting immediately? It seems like we would not have any responses anyway, so it should happen pretty quickly either way (and all pending requests have been cancelled). Maybe keep it simple?

ijuma · 2017-09-21T07:53:54Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

@@ -579,6 +580,11 @@ synchronized void retry(TxnRequestHandler request) {
        enqueueRequest(request);
    }

+    synchronized void authenticationFailed(AuthenticationException e) {


Would it be worth making this more generic and simply call it fatalError or something?

Hmm... The code was changed from generic to more specific in the other PR based on review comments. So I thought this should be specific too.

OK, we can leave it like that then.

ijuma · 2017-09-21T07:58:20Z

clients/src/main/java/org/apache/kafka/common/errors/IllegalSaslStateException.java

+ * This exception indicates unexpected requests prior to SASL authentication.
+ * This could be due to misconfigured security, e.g. if PLAINTEXT protocol
+ * is used to connect to a SASL endpoint.
+ *


Nit: empty line.

ijuma · 2017-09-21T07:58:26Z

clients/src/main/java/org/apache/kafka/common/errors/UnsupportedSaslMechanismException.java

+/**
+ * This exception indicates that the SASL mechanism requested by the client
+ * is not enabled on the broker.
+ *


Nit: empty line.

ijuma · 2017-09-21T07:59:13Z

...est/java/org/apache/kafka/common/security/authenticator/ClientAuthenticationFailureTest.java

+            result.all().get();
+            fail("Expected an authentication error!");
+        } catch (Exception e) {
+            assertTrue("Expected AuthenticationFailedException, got " + e.getClass(), e.getCause() instanceof AuthenticationFailedException);


Nit: long line.

ijuma · 2017-09-21T08:12:44Z

...est/java/org/apache/kafka/common/security/authenticator/ClientAuthenticationFailureTest.java

+        Map<String, Object> props = new HashMap<>(saslClientConfigs);
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:" + server.port());
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");


We can pass the serializers in the constructor and it's a bit more concise.

To add to the constructors, we need to remove the import restrictions on the package. It feels better to do them as properties instead since this test is currently in the security package. @vahidhashemian did that to avoid changing the restrictions.

The serialization package is open to everyone (it's public API and at the lowest layer). So I don't think we should worry about that. It's not like we're avoiding a dependency here, we are just hiding it via a string based config (that still requires the default constructor to be present).

ijuma · 2017-09-21T08:13:02Z

...est/java/org/apache/kafka/common/security/authenticator/ClientAuthenticationFailureTest.java

+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "txclient-1");
+        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5");


This is the default.

copied and pasted earlier :-), removed now.

ijuma · 2017-09-21T08:13:20Z

...est/java/org/apache/kafka/common/security/authenticator/ClientAuthenticationFailureTest.java

+        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "txclient-1");
+        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5");
+        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
+        props.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");


This is also the default, I think.

ijuma · 2017-09-21T08:20:45Z

core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala

+    producerConfig.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "txclient-1")
+    producerConfig.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5")
+    producerConfig.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true")
+    producerConfig.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384")


This and max in flight are defaults right?

yes, removed.

ijuma

Thanks for the updates, a few more comments.

ijuma · 2017-09-21T09:55:08Z

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

+        private boolean handleAuthenticationException(long now, Map<Node, List<Call>> callsToSend) {
+            AuthenticationException authenticationException = null;
+            try {
+                metadata.maybeThrowAuthenticationException();


Should we add a method that is like this one, but returns the exception instead? It seems a bit convoluted to throw a stored exception just to catch it. Not sure if that helps other cases, but at least here, it seems to make the code cleaner.

I changed the existing method to return the exception rather than throw it.

ijuma · 2017-09-21T10:00:32Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

@@ -579,6 +580,11 @@ synchronized void retry(TxnRequestHandler request) {
        enqueueRequest(request);
    }

+    synchronized void authenticationFailed(AuthenticationException e) {


OK, we can leave it like that then.

ijuma · 2017-09-21T10:01:44Z

...est/java/org/apache/kafka/common/security/authenticator/ClientAuthenticationFailureTest.java

+        Map<String, Object> props = new HashMap<>(saslClientConfigs);
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:" + server.port());
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");


The serialization package is open to everyone (it's public API and at the lowest layer). So I don't think we should worry about that. It's not like we're avoiding a dependency here, we are just hiding it via a string based config (that still requires the default constructor to be present).

ijuma · 2017-09-21T10:17:48Z

core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala

@@ -68,12 +74,27 @@ class SaslClientsWithInvalidCredentialsTest extends IntegrationTestHarness with
  @Test
  def testProducerWithAuthenticationFailure() {
    verifyAuthenticationException(() => sendOneRecord(10000))
+    verifyAuthenticationException(() => producers.head.partitionsFor(topic))


Nit: we can change the parameter to be by name (=> Unit instead of () => Unit) and then you don't need the noisy () => in every invocation.

ijuma · 2017-09-21T10:18:33Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

+                }
+            } catch (AuthenticationException e) {
+                // This is already logged as error, but propagated here to perform any clean ups.
+                log.trace("Authentication exception while processing transactional request");


Do we want to include the exception message or stacktrace?

ijuma · 2017-09-21T10:30:53Z

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

+                }
+                for (List<Call> calls : callsToSend.values()) {
+                    for (Call call : calls) {
+                        call.handleFailure(authenticationException);


Why are we calling handleFailure directly instead of fail? If we used fail in both places, we could extract a method and call it twice in succession, once for newCalls and once for callsToSend.

ijuma · 2017-09-21T10:36:58Z

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -974,6 +1011,9 @@ public void run() {

                // Update the current time and handle the latest responses.
                now = time.milliseconds();
+                if (handleAuthenticationException(now, callsToSend) &&
+                        hardShutdownTimeMs.get() != INVALID_SHUTDOWN_TIME)


Hmm, what's the impact of not exiting immediately? It seems like we would not have any responses anyway, so it should happen pretty quickly either way (and all pending requests have been cancelled). Maybe keep it simple?

rajinisivaram · 2017-09-21T11:27:51Z

@vahidhashemian @ijuma Thanks for the reviews. Have addressed the comments so far.

ijuma

LGTM. I'll merge it to trunk once the tests pass. If there are additional comments, we can do a follow-up.

ijuma · 2017-09-21T13:03:39Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

@@ -200,32 +201,38 @@ public void run() {
     */
    void run(long now) {
        if (transactionManager != null) {
-            if (transactionManager.shouldResetProducerStateAfterResolvingSequences())


@hachikuji, if you have a chance, have a look and see if this is OK.

Looks ok to me.

KAFKA-5947: Handle authentication exceptions in admin client and txn

764f387

producer

ijuma reviewed Sep 21, 2017

View reviewed changes

Address review comments

362ab33

ijuma reviewed Sep 21, 2017

View reviewed changes

Address review comments

4143990

ijuma approved these changes Sep 21, 2017

View reviewed changes

asfgit closed this in 96ba21e Sep 21, 2017

ijuma reviewed Sep 21, 2017

View reviewed changes

KAFKA-5947: Handle authentication failure in admin client, txn producer #3928

KAFKA-5947: Handle authentication failure in admin client, txn producer #3928

Conversation

rajinisivaram commented Sep 20, 2017

rajinisivaram commented Sep 20, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rajinisivaram Sep 21, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vahidhashemian commented Sep 21, 2017

ijuma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ijuma Sep 21, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rajinisivaram commented Sep 21, 2017

ijuma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rajinisivaram Sep 21, 2017 •

edited

Loading

ijuma Sep 21, 2017 •

edited

Loading