kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jun...@apache.org
Subject kafka git commit: KAFKA-2847; Remove principal builder class from client configs
Date Tue, 17 Nov 2015 16:36:59 GMT
Repository: kafka
Updated Branches:
  refs/heads/0.9.0 cae36f5c3 -> 154073026


KAFKA-2847; Remove principal builder class from client configs

Also mark `PrincipalBuilder` as `Unstable` and  tweak docs.

Author: Ismael Juma <ismael@juma.me.uk>

Reviewers: Jun Rao <junrao@gmail.com>

Closes #542 from ijuma/kafka-2847-remove-principal-builder-class-from-client-configs

(cherry picked from commit 52d5e88393630ce6f817bd003c7c787e36e31277)
Signed-off-by: Jun Rao <junrao@gmail.com>


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

Branch: refs/heads/0.9.0
Commit: 15407302621a2d827bddbe8085759f6f74ff786d
Parents: cae36f5
Author: Ismael Juma <ismael@juma.me.uk>
Authored: Tue Nov 17 08:36:43 2015 -0800
Committer: Jun Rao <junrao@gmail.com>
Committed: Tue Nov 17 08:36:52 2015 -0800

----------------------------------------------------------------------
 checkstyle/import-control.xml                    |  1 +
 .../apache/kafka/common/config/SslConfigs.java   |  7 ++++---
 .../kafka/common/network/ChannelBuilders.java    | 19 +++++++++++++++++++
 .../common/network/PlaintextChannelBuilder.java  |  5 +----
 .../kafka/common/network/SaslChannelBuilder.java | 10 ++--------
 .../kafka/common/network/SslChannelBuilder.java  |  5 +----
 .../common/security/auth/PrincipalBuilder.java   |  2 ++
 .../clients/producer/KafkaProducerTest.java      |  4 +---
 .../kafka/common/network/SelectorTest.java       |  4 +---
 .../kafka/common/network/SslSelectorTest.java    |  1 -
 .../common/network/SslTransportLayerTest.java    |  5 +++--
 11 files changed, 35 insertions(+), 28 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/15407302/checkstyle/import-control.xml
----------------------------------------------------------------------
diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml
index 4fdf95a..588ceff 100644
--- a/checkstyle/import-control.xml
+++ b/checkstyle/import-control.xml
@@ -66,6 +66,7 @@
     </subpackage>
 
     <subpackage name="security">
+      <allow pkg="org.apache.kafka.common.annotation" />
       <allow pkg="org.apache.kafka.common.network" />
       <allow pkg="org.apache.kafka.common.config" />
     </subpackage>

http://git-wip-us.apache.org/repos/asf/kafka/blob/15407302/clients/src/main/java/org/apache/kafka/common/config/SslConfigs.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/config/SslConfigs.java b/clients/src/main/java/org/apache/kafka/common/config/SslConfigs.java
index ae4667a..a893b75 100644
--- a/clients/src/main/java/org/apache/kafka/common/config/SslConfigs.java
+++ b/clients/src/main/java/org/apache/kafka/common/config/SslConfigs.java
@@ -22,7 +22,9 @@ public class SslConfigs {
      */
 
     public static final String PRINCIPAL_BUILDER_CLASS_CONFIG = "principal.builder.class";
-    public static final String PRINCIPAL_BUILDER_CLASS_DOC = "principal builder to generate
a java Principal. This config is optional for client.";
+    public static final String PRINCIPAL_BUILDER_CLASS_DOC = "The fully qualified name of
a class that implements the PrincipalBuilder interface, " +
+            "which is currently used to build the Principal for connections with the SSL
SecurityProtocol. " +
+            "Default is DefaultPrincipalBuilder.";
     public static final String DEFAULT_PRINCIPAL_BUILDER_CLASS = "org.apache.kafka.common.security.auth.DefaultPrincipalBuilder";
 
     public static final String SSL_PROTOCOL_CONFIG = "ssl.protocol";
@@ -97,8 +99,7 @@ public class SslConfigs {
                                            + " <li><code>ssl.client.auth=none</code>
This means client authentication is not needed.";
 
     public static void addClientSslSupport(ConfigDef config) {
-        config.define(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, ConfigDef.Type.CLASS, SslConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS,
ConfigDef.Importance.LOW, SslConfigs.PRINCIPAL_BUILDER_CLASS_DOC)
-                .define(SslConfigs.SSL_PROTOCOL_CONFIG, ConfigDef.Type.STRING, SslConfigs.DEFAULT_SSL_PROTOCOL,
ConfigDef.Importance.MEDIUM, SslConfigs.SSL_PROTOCOL_DOC)
+        config.define(SslConfigs.SSL_PROTOCOL_CONFIG, ConfigDef.Type.STRING, SslConfigs.DEFAULT_SSL_PROTOCOL,
ConfigDef.Importance.MEDIUM, SslConfigs.SSL_PROTOCOL_DOC)
                 .define(SslConfigs.SSL_PROVIDER_CONFIG, ConfigDef.Type.STRING, null, ConfigDef.Importance.MEDIUM,
SslConfigs.SSL_PROVIDER_DOC)
                 .define(SslConfigs.SSL_CIPHER_SUITES_CONFIG, ConfigDef.Type.LIST, null, ConfigDef.Importance.LOW,
SslConfigs.SSL_CIPHER_SUITES_DOC)
                 .define(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, ConfigDef.Type.LIST, SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS,
ConfigDef.Importance.MEDIUM, SslConfigs.SSL_ENABLED_PROTOCOLS_DOC)

http://git-wip-us.apache.org/repos/asf/kafka/blob/15407302/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java b/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java
index 03c663d..669f269 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java
@@ -13,7 +13,11 @@
 
 package org.apache.kafka.common.network;
 
+import org.apache.kafka.common.config.SslConfigs;
 import org.apache.kafka.common.protocol.SecurityProtocol;
+import org.apache.kafka.common.security.auth.DefaultPrincipalBuilder;
+import org.apache.kafka.common.security.auth.PrincipalBuilder;
+import org.apache.kafka.common.utils.Utils;
 
 import java.util.Map;
 
@@ -57,6 +61,21 @@ public class ChannelBuilders {
         return channelBuilder;
     }
 
+    /**
+     * Returns a configured `PrincipalBuilder`.
+     */
+    static PrincipalBuilder createPrincipalBuilder(Map<String, ?> configs) {
+        // this is a server-only config so it will always be null on the client
+        Class<?> principalBuilderClass = (Class<?>) configs.get(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG);
+        PrincipalBuilder principalBuilder;
+        if (principalBuilderClass == null)
+            principalBuilder = new DefaultPrincipalBuilder();
+        else
+            principalBuilder = (PrincipalBuilder) Utils.newInstance(principalBuilderClass);
+        principalBuilder.configure(configs);
+        return principalBuilder;
+    }
+
     private static void requireNonNullMode(Mode mode, SecurityProtocol securityProtocol)
{
         if (mode == null)
             throw new IllegalArgumentException("`mode` must be non-null if `securityProtocol`
is `" + securityProtocol + "`");

http://git-wip-us.apache.org/repos/asf/kafka/blob/15407302/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java
b/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java
index bc1536a..f0af935 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/PlaintextChannelBuilder.java
@@ -15,9 +15,7 @@ package org.apache.kafka.common.network;
 import java.nio.channels.SelectionKey;
 import java.util.Map;
 
-import org.apache.kafka.common.config.SslConfigs;
 import org.apache.kafka.common.security.auth.PrincipalBuilder;
-import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.common.KafkaException;
 
 import org.slf4j.Logger;
@@ -31,8 +29,7 @@ public class PlaintextChannelBuilder implements ChannelBuilder {
     public void configure(Map<String, ?> configs) throws KafkaException {
         try {
             this.configs = configs;
-            this.principalBuilder = (PrincipalBuilder) Utils.newInstance((Class<?>)
configs.get(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG));
-            this.principalBuilder.configure(this.configs);
+            principalBuilder = ChannelBuilders.createPrincipalBuilder(configs);
         } catch (Exception e) {
             throw new KafkaException(e);
         }

http://git-wip-us.apache.org/repos/asf/kafka/blob/15407302/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
b/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
index 75e3fca..86ac779 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
@@ -20,15 +20,12 @@ import java.util.Map;
 
 import org.apache.kafka.common.config.SaslConfigs;
 import org.apache.kafka.common.security.JaasUtils;
-import org.apache.kafka.common.security.auth.PrincipalBuilder;
 import org.apache.kafka.common.security.kerberos.KerberosShortNamer;
 import org.apache.kafka.common.security.kerberos.LoginManager;
 import org.apache.kafka.common.security.authenticator.SaslClientAuthenticator;
 import org.apache.kafka.common.security.authenticator.SaslServerAuthenticator;
 import org.apache.kafka.common.security.ssl.SslFactory;
 import org.apache.kafka.common.protocol.SecurityProtocol;
-import org.apache.kafka.common.config.SslConfigs;
-import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.common.KafkaException;
 
 import org.slf4j.Logger;
@@ -42,7 +39,6 @@ public class SaslChannelBuilder implements ChannelBuilder {
     private final LoginType loginType;
 
     private LoginManager loginManager;
-    private PrincipalBuilder principalBuilder;
     private SslFactory sslFactory;
     private Map<String, ?> configs;
     private KerberosShortNamer kerberosShortNamer;
@@ -57,8 +53,6 @@ public class SaslChannelBuilder implements ChannelBuilder {
         try {
             this.configs = configs;
             this.loginManager = LoginManager.acquireLoginManager(loginType, configs);
-            this.principalBuilder = (PrincipalBuilder) Utils.newInstance((Class<?>)
configs.get(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG));
-            this.principalBuilder.configure(configs);
 
             String defaultRealm;
             try {
@@ -90,7 +84,8 @@ public class SaslChannelBuilder implements ChannelBuilder {
             else
                 authenticator = new SaslClientAuthenticator(id, loginManager.subject(), loginManager.serviceName(),
                         socketChannel.socket().getInetAddress().getHostName());
-            authenticator.configure(transportLayer, this.principalBuilder, this.configs);
+            // Both authenticators don't use `PrincipalBuilder`, so we pass `null` for now.
Reconsider if this changes.
+            authenticator.configure(transportLayer, null, this.configs);
             return new KafkaChannel(id, transportLayer, authenticator, maxReceiveSize);
         } catch (Exception e) {
             log.info("Failed to create channel due to ", e);
@@ -99,7 +94,6 @@ public class SaslChannelBuilder implements ChannelBuilder {
     }
 
     public void close()  {
-        this.principalBuilder.close();
         this.loginManager.release();
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/15407302/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
b/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
index 9a7ba0c..b546174 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
@@ -19,8 +19,6 @@ import java.util.Map;
 
 import org.apache.kafka.common.security.auth.PrincipalBuilder;
 import org.apache.kafka.common.security.ssl.SslFactory;
-import org.apache.kafka.common.config.SslConfigs;
-import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.common.KafkaException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -41,8 +39,7 @@ public class SslChannelBuilder implements ChannelBuilder {
             this.configs = configs;
             this.sslFactory = new SslFactory(mode);
             this.sslFactory.configure(this.configs);
-            this.principalBuilder = (PrincipalBuilder) Utils.newInstance((Class<?>)
configs.get(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG));
-            this.principalBuilder.configure(this.configs);
+            this.principalBuilder = ChannelBuilders.createPrincipalBuilder(configs);
         } catch (Exception e) {
             throw new KafkaException(e);
         }

http://git-wip-us.apache.org/repos/asf/kafka/blob/15407302/clients/src/main/java/org/apache/kafka/common/security/auth/PrincipalBuilder.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/common/security/auth/PrincipalBuilder.java
b/clients/src/main/java/org/apache/kafka/common/security/auth/PrincipalBuilder.java
index 99b6d21..75e1855 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/auth/PrincipalBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/auth/PrincipalBuilder.java
@@ -17,6 +17,7 @@
 
 package org.apache.kafka.common.security.auth;
 
+import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.common.network.TransportLayer;
 import org.apache.kafka.common.network.Authenticator;
 import org.apache.kafka.common.KafkaException;
@@ -28,6 +29,7 @@ import java.security.Principal;
 /*
  * PrincipalBuilder for Authenticator
  */
+@InterfaceStability.Unstable
 public interface PrincipalBuilder extends Configurable {
 
     /**

http://git-wip-us.apache.org/repos/asf/kafka/blob/15407302/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
index b96a5f7..1130225 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
@@ -18,7 +18,6 @@ package org.apache.kafka.clients.producer;
 
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.apache.kafka.common.config.SslConfigs;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.test.MockMetricsReporter;
 import org.apache.kafka.test.MockSerializer;
@@ -54,12 +53,11 @@ public class KafkaProducerTest {
 
     @Test
     public void testSerializerClose() throws Exception {
-        Map<String, Object> configs = new HashMap<String, Object>();
+        Map<String, Object> configs = new HashMap<>();
         configs.put(ProducerConfig.CLIENT_ID_CONFIG, "testConstructorClose");
         configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
         configs.put(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, MockMetricsReporter.class.getName());
         configs.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL);
-        configs.put(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, Class.forName(SslConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS));
         final int oldInitCount = MockSerializer.INIT_COUNT.get();
         final int oldCloseCount = MockSerializer.CLOSE_COUNT.get();
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/15407302/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
index 8ce0298..18fd080 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
@@ -24,7 +24,6 @@ import java.nio.ByteBuffer;
 import java.util.*;
 
 import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.config.SslConfigs;
 import org.apache.kafka.common.utils.MockTime;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Utils;
@@ -48,8 +47,7 @@ public class SelectorTest {
 
     @Before
     public void setup() throws Exception {
-        Map<String, Object> configs = new HashMap<String, Object>();
-        configs.put(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, Class.forName(SslConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS));
+        Map<String, Object> configs = new HashMap<>();
         this.server = new EchoServer(configs);
         this.server.start();
         this.time = new MockTime();

http://git-wip-us.apache.org/repos/asf/kafka/blob/15407302/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
index 94c5654..a442ea0 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
@@ -52,7 +52,6 @@ public class SslSelectorTest extends SelectorTest {
         this.server.start();
         this.time = new MockTime();
         sslClientConfigs = TestSslUtils.createSslConfig(false, false, Mode.SERVER, trustStoreFile,
"client");
-        sslClientConfigs.put(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, Class.forName(SslConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS));
 
         this.channelBuilder = new SslChannelBuilder(Mode.CLIENT);
         this.channelBuilder.configure(sslClientConfigs);

http://git-wip-us.apache.org/repos/asf/kafka/blob/15407302/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
b/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
index 282ff8b..2b5d26b 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
@@ -464,11 +464,12 @@ public class SslTransportLayerTest {
             Mode mode = server ? Mode.SERVER : Mode.CLIENT;
             File truststoreFile = File.createTempFile(name + "TS", ".jks");
             sslConfig = TestSslUtils.createSslConfig(!server, true, mode, truststoreFile,
name);
-            sslConfig.put(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, Class.forName(SslConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS));
+            if (server)
+                sslConfig.put(SslConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG, Class.forName(SslConfigs.DEFAULT_PRINCIPAL_BUILDER_CLASS));
         }
        
         private Map<String, Object> getTrustingConfig(CertStores truststoreConfig)
{
-            Map<String, Object> config = new HashMap<String, Object>(sslConfig);
+            Map<String, Object> config = new HashMap<>(sslConfig);
             config.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, truststoreConfig.sslConfig.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG));
             config.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, truststoreConfig.sslConfig.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG));
             config.put(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, truststoreConfig.sslConfig.get(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG));


Mime
View raw message