kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rsiva...@apache.org
Subject kafka git commit: MINOR: Fix unit test failures from last commit, cherry-pick fix for SASL principal
Date Thu, 31 Aug 2017 20:20:58 GMT
Repository: kafka
Updated Branches:
  refs/heads/0.10.2 0b4daa4bf -> c9168e10f


MINOR: Fix unit test failures from last commit, cherry-pick fix for SASL principal

This fixes PlainSaslServerTest from the last commit which uses a class not present in 0.10.2.
Also cherry-picked commit 9934d28 to 0.10.2 since that is required for multiple mechanism
support (SaslMultiMechanismConsumerTest was failing with the changes without this).

Author: Rajini Sivaram <rajinisivaram@googlemail.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>, Jun Rao <junrao@gmail.com>

Closes #3773 from rajinisivaram/MINOR-0.10.2-buildfix


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

Branch: refs/heads/0.10.2
Commit: c9168e10f93d713ad78ed421f50e9819b3f046b5
Parents: 0b4daa4
Author: Rajini Sivaram <rajinisivaram@googlemail.com>
Authored: Thu Aug 31 16:20:33 2017 -0400
Committer: Rajini Sivaram <rajinisivaram@googlemail.com>
Committed: Thu Aug 31 16:20:33 2017 -0400

----------------------------------------------------------------------
 .../consumer/internals/AbstractCoordinator.java |  2 +-
 .../apache/kafka/common/network/Selector.java   | 17 +++++++++-
 .../kafka/common/protocol/types/Struct.java     |  2 +-
 .../authenticator/SaslClientAuthenticator.java  | 35 ++++++++++++++++----
 .../SaslClientCallbackHandler.java              |  2 +-
 .../authenticator/SaslServerAuthenticator.java  | 10 +++---
 .../clients/producer/ProducerRecordTest.java    | 12 +++----
 .../authenticator/SaslAuthenticatorTest.java    |  9 +++--
 .../security/plain/PlainSaslServerTest.java     | 16 ++++++---
 .../security/scram/ScramSaslServerTest.java     |  2 +-
 10 files changed, 76 insertions(+), 31 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/c9168e10/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 d0d5e8d..c8096ef 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
@@ -966,7 +966,7 @@ public abstract class AbstractCoordinator implements Closeable {
                 log.error("Unexpected interrupt received in heartbeat thread for group {}",
groupId, e);
                 this.failed.set(new RuntimeException(e));
             } catch (RuntimeException e) {
-                log.error("Heartbeat thread for group {} failed due to unexpected error"
, groupId, e);
+                log.error("Heartbeat thread for group {} failed due to unexpected error",
groupId, e);
                 this.failed.set(e);
             } finally {
                 log.debug("Heartbeat thread for group {} has closed", groupId);

http://git-wip-us.apache.org/repos/asf/kafka/blob/c9168e10/clients/src/main/java/org/apache/kafka/common/network/Selector.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/Selector.java b/clients/src/main/java/org/apache/kafka/common/network/Selector.java
index df35266..f55dc96 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/Selector.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/Selector.java
@@ -185,7 +185,17 @@ public class Selector implements Selectable {
             throw e;
         }
         SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_CONNECT);
-        KafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize);
+        KafkaChannel channel;
+        try {
+            channel = channelBuilder.buildChannel(id, key, maxReceiveSize);
+        } catch (Exception e) {
+            try {
+                socketChannel.close();
+            } finally {
+                key.cancel();
+            }
+            throw new IOException("Channel could not be created for socket " + socketChannel,
e);
+        }
         key.attach(channel);
         this.channels.put(id, channel);
 
@@ -650,6 +660,11 @@ public class Selector implements Selectable {
         this.sensors.recordBytesReceived(channel.id(), networkReceive.payload().limit());
     }
 
+    // only for testing
+    public Set<SelectionKey> keys() {
+        return new HashSet<>(nioSelector.keys());
+    }
+
     private class SelectorMetrics {
         private final Metrics metrics;
         public final Sensor connectionClosed;

http://git-wip-us.apache.org/repos/asf/kafka/blob/c9168e10/clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java b/clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java
index 0165ce6..ea5e948 100644
--- a/clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java
+++ b/clients/src/main/java/org/apache/kafka/common/protocol/types/Struct.java
@@ -312,7 +312,7 @@ public class Struct {
             Field f = this.schema.get(i);
             if (f.type() instanceof ArrayOf) {
                 if (this.get(f) != null) {
-                    Object[] arrayObject = (Object []) this.get(f);
+                    Object[] arrayObject = (Object[]) this.get(f);
                     for (Object arrayItem: arrayObject)
                         result = prime * result + arrayItem.hashCode();
                 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/c9168e10/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java
b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java
index 59eee83..dd793b6 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java
@@ -21,6 +21,7 @@ package org.apache.kafka.common.security.authenticator;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.NetworkClient;
 import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.errors.IllegalSaslStateException;
 import org.apache.kafka.common.errors.UnsupportedSaslMechanismException;
@@ -54,7 +55,9 @@ import java.security.Principal;
 import java.security.PrivilegedActionException;
 import java.security.PrivilegedExceptionAction;
 import java.util.Arrays;
+import java.util.Iterator;
 import java.util.Map;
+import java.util.Set;
 
 public class SaslClientAuthenticator implements Authenticator {
 
@@ -108,13 +111,14 @@ public class SaslClientAuthenticator implements Authenticator {
 
             setSaslState(handshakeRequestEnable ? SaslState.SEND_HANDSHAKE_REQUEST : SaslState.INITIAL);
 
-            // determine client principal from subject.
-            if (!subject.getPrincipals().isEmpty()) {
-                Principal clientPrincipal = subject.getPrincipals().iterator().next();
-                this.clientPrincipalName = clientPrincipal.getName();
-            } else {
-                clientPrincipalName = null;
-            }
+            // determine client principal from subject for Kerberos to use as authorization
id for the SaslClient.
+            // For other mechanisms, the authenticated principal (username for PLAIN and
SCRAM) is used as
+            // authorization id. Hence the principal is not specified for creating the SaslClient.
+            if (mechanism.equals(SaslConfigs.GSSAPI_MECHANISM))
+                this.clientPrincipalName = firstPrincipal(subject);
+            else
+                this.clientPrincipalName = null;
+
             callbackHandler = new SaslClientCallbackHandler();
             callbackHandler.configure(configs, Mode.CLIENT, subject, mechanism);
 
@@ -339,4 +343,21 @@ public class SaslClientAuthenticator implements Authenticator {
                     response.errorCode(), mechanism, response.enabledMechanisms()));
         }
     }
+
+    /**
+     * Returns the first Principal from Subject.
+     * @throws KafkaException if there are no Principals in the Subject.
+     *     During Kerberos re-login, principal is reset on Subject. An exception is
+     *     thrown so that the connection is retried after any configured backoff.
+     */
+    static final String firstPrincipal(Subject subject) {
+        Set<Principal> principals = subject.getPrincipals();
+        synchronized (principals) {
+            Iterator<Principal> iterator = principals.iterator();
+            if (iterator.hasNext())
+                return iterator.next().getName();
+            else
+                throw new KafkaException("Principal could not be determined from Subject,
this may be a transient failure due to Kerberos re-login");
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/c9168e10/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientCallbackHandler.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientCallbackHandler.java
b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientCallbackHandler.java
index 8e0b8db..6094b54 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientCallbackHandler.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientCallbackHandler.java
@@ -59,7 +59,7 @@ public class SaslClientCallbackHandler implements AuthCallbackHandler {
                     nc.setName(nc.getDefaultName());
             } else if (callback instanceof PasswordCallback) {
                 if (!isKerberos && subject != null && !subject.getPrivateCredentials(String.class).isEmpty())
{
-                    char [] password = subject.getPrivateCredentials(String.class).iterator().next().toCharArray();
+                    char[] password = subject.getPrivateCredentials(String.class).iterator().next().toCharArray();
                     ((PasswordCallback) callback).setPassword(password);
                 } else {
                     String errorMessage = "Could not login: the client is being asked for
a password, but the Kafka" +

http://git-wip-us.apache.org/repos/asf/kafka/blob/c9168e10/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
index 069e12f..c71802e 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java
@@ -134,9 +134,7 @@ public class SaslServerAuthenticator implements Authenticator {
             callbackHandler = new ScramServerCallbackHandler(credentialCache.cache(mechanism,
ScramCredential.class));
         callbackHandler.configure(configs, Mode.SERVER, subject, saslMechanism);
         if (mechanism.equals(SaslConfigs.GSSAPI_MECHANISM)) {
-            if (subject.getPrincipals().isEmpty())
-                throw new IllegalArgumentException("subject must have at least one principal");
-            saslServer = createSaslKerberosServer(callbackHandler, configs);
+            saslServer = createSaslKerberosServer(callbackHandler, configs, subject);
         } else {
             try {
                 saslServer = Subject.doAs(subject, new PrivilegedExceptionAction<SaslServer>()
{
@@ -150,12 +148,12 @@ public class SaslServerAuthenticator implements Authenticator {
         }
     }
 
-    private SaslServer createSaslKerberosServer(final AuthCallbackHandler saslServerCallbackHandler,
final Map<String, ?> configs) throws IOException {
+    private SaslServer createSaslKerberosServer(final AuthCallbackHandler saslServerCallbackHandler,
final Map<String, ?> configs, Subject subject) throws IOException {
         // server is using a JAAS-authenticated subject: determine service principal name
and hostname from kafka server's subject.
-        final Principal servicePrincipal = subject.getPrincipals().iterator().next();
+        final String servicePrincipal = SaslClientAuthenticator.firstPrincipal(subject);
         KerberosName kerberosName;
         try {
-            kerberosName = KerberosName.parse(servicePrincipal.getName());
+            kerberosName = KerberosName.parse(servicePrincipal);
         } catch (IllegalArgumentException e) {
             throw new KafkaException("Principal has name with unexpected format " + servicePrincipal);
         }

http://git-wip-us.apache.org/repos/asf/kafka/blob/c9168e10/clients/src/test/java/org/apache/kafka/clients/producer/ProducerRecordTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/ProducerRecordTest.java
b/clients/src/test/java/org/apache/kafka/clients/producer/ProducerRecordTest.java
index a844bb0..5186d05 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/ProducerRecordTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/ProducerRecordTest.java
@@ -26,24 +26,24 @@ public class ProducerRecordTest {
 
     @Test
     public void testEqualsAndHashCode() {
-        ProducerRecord<String, Integer> producerRecord = new ProducerRecord<>("test",
1 , "key", 1);
+        ProducerRecord<String, Integer> producerRecord = new ProducerRecord<>("test",
1, "key", 1);
         assertEquals(producerRecord, producerRecord);
         assertEquals(producerRecord.hashCode(), producerRecord.hashCode());
 
-        ProducerRecord<String, Integer> equalRecord = new ProducerRecord<>("test",
1 , "key", 1);
+        ProducerRecord<String, Integer> equalRecord = new ProducerRecord<>("test",
1, "key", 1);
         assertEquals(producerRecord, equalRecord);
         assertEquals(producerRecord.hashCode(), equalRecord.hashCode());
 
-        ProducerRecord<String, Integer> topicMisMatch = new ProducerRecord<>("test-1",
1 , "key", 1);
+        ProducerRecord<String, Integer> topicMisMatch = new ProducerRecord<>("test-1",
1, "key", 1);
         assertFalse(producerRecord.equals(topicMisMatch));
 
-        ProducerRecord<String, Integer> partitionMismatch = new ProducerRecord<>("test",
2 , "key", 1);
+        ProducerRecord<String, Integer> partitionMismatch = new ProducerRecord<>("test",
2, "key", 1);
         assertFalse(producerRecord.equals(partitionMismatch));
 
-        ProducerRecord<String, Integer> keyMisMatch = new ProducerRecord<>("test",
1 , "key-1", 1);
+        ProducerRecord<String, Integer> keyMisMatch = new ProducerRecord<>("test",
1, "key-1", 1);
         assertFalse(producerRecord.equals(keyMisMatch));
 
-        ProducerRecord<String, Integer> valueMisMatch = new ProducerRecord<>("test",
1 , "key", 2);
+        ProducerRecord<String, Integer> valueMisMatch = new ProducerRecord<>("test",
1, "key", 2);
         assertFalse(producerRecord.equals(valueMisMatch));
 
         ProducerRecord<String, Integer> nullFieldsRecord = new ProducerRecord<>("topic",
null, null, null, null);

http://git-wip-us.apache.org/repos/asf/kafka/blob/c9168e10/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
index ac9beb4..c8335aa 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
@@ -51,6 +51,7 @@ import org.junit.Test;
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
 import java.security.NoSuchAlgorithmException;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -63,6 +64,7 @@ import java.util.Random;
 import javax.security.auth.login.Configuration;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
@@ -169,8 +171,11 @@ public class SaslAuthenticatorTest {
         try {
             selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE);
             fail("SASL/PLAIN channel created without username");
-        } catch (KafkaException e) {
+        } catch (IOException e) {
             // Expected exception
+            assertTrue("Channels not closed", selector.channels().isEmpty());
+            for (SelectionKey key : selector.keys())
+                assertFalse("Key not cancelled", key.isValid());
         }
     }
 
@@ -190,7 +195,7 @@ public class SaslAuthenticatorTest {
         try {
             selector.connect(node, addr, BUFFER_SIZE, BUFFER_SIZE);
             fail("SASL/PLAIN channel created without password");
-        } catch (KafkaException e) {
+        } catch (IOException e) {
             // Expected exception
         }
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/c9168e10/clients/src/test/java/org/apache/kafka/common/security/plain/PlainSaslServerTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/plain/PlainSaslServerTest.java
b/clients/src/test/java/org/apache/kafka/common/security/plain/PlainSaslServerTest.java
index 5ca778b..039e385 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/plain/PlainSaslServerTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/plain/PlainSaslServerTest.java
@@ -1,4 +1,4 @@
-/*
+/**
  * 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.
@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.common.security.plain;
 
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -23,11 +24,11 @@ import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.Map;
 
+import javax.security.auth.login.Configuration;
 import javax.security.sasl.SaslException;
 
 import static org.junit.Assert.assertEquals;
 
-import org.apache.kafka.common.security.JaasContext;
 import org.apache.kafka.common.security.authenticator.TestJaasConfig;
 
 public class PlainSaslServerTest {
@@ -45,9 +46,14 @@ public class PlainSaslServerTest {
         Map<String, Object> options = new HashMap<>();
         options.put("user_" + USER_A, PASSWORD_A);
         options.put("user_" + USER_B, PASSWORD_B);
-        jaasConfig.addEntry("jaasContext", PlainLoginModule.class.getName(), options);
-        JaasContext jaasContext = new JaasContext("jaasContext", JaasContext.Type.SERVER,
jaasConfig);
-        saslServer = new PlainSaslServer(jaasContext);
+        jaasConfig.addEntry("KafkaServer", PlainLoginModule.class.getName(), options);
+        Configuration.setConfiguration(jaasConfig);
+        saslServer = new PlainSaslServer(null);
+    }
+
+    @After
+    public void tearDown() {
+        Configuration.setConfiguration(null);
     }
 
     @Test

http://git-wip-us.apache.org/repos/asf/kafka/blob/c9168e10/clients/src/test/java/org/apache/kafka/common/security/scram/ScramSaslServerTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/security/scram/ScramSaslServerTest.java
b/clients/src/test/java/org/apache/kafka/common/security/scram/ScramSaslServerTest.java
index 9b329d0..a1d51f0 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/scram/ScramSaslServerTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/scram/ScramSaslServerTest.java
@@ -1,4 +1,4 @@
-/*
+/**
  * 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.


Mime
View raw message