[Issue 4283][flink] construct auth when building pulsar source (apach…

…e#4284) Fixes apache#4283 ### Motivation pulsar flink connector now uses ClientConfigData to instantiate pulsar client. Pulsar client composes Authentication which can not be serialized. In an environment with Auth there is no way to set auth in pulsar client. ### Modifications Keep auth params away from persistence and serialization. Construct auth when building pulsar source
AlphaWang · May 18, 2019 · 353ca73 · 353ca73
1 parent ce685dc
commit 353ca73
Show file tree

Hide file tree

Showing 4 changed files with 42 additions and 19 deletions.
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ClientConfigurationData.java
@@ -18,19 +18,18 @@
  */
 package org.apache.pulsar.client.impl.conf;
 
-import java.io.Serializable;
-import java.util.concurrent.TimeUnit;
-
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
+import lombok.Data;
 import lombok.NoArgsConstructor;
 import org.apache.pulsar.client.api.Authentication;
 import org.apache.pulsar.client.api.ServiceUrlProvider;
 import org.apache.pulsar.client.impl.auth.AuthenticationDisabled;
 
-import com.fasterxml.jackson.annotation.JsonIgnore;
-
-import lombok.Data;
+import java.io.Serializable;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 /**
  * This is a simple holder of the client configuration values.
@@ -44,10 +43,15 @@ public class ClientConfigurationData implements Serializable, Cloneable {
 
     private String serviceUrl;
     @JsonIgnore
-    private ServiceUrlProvider serviceUrlProvider;
+    private transient ServiceUrlProvider serviceUrlProvider;
 
     @JsonIgnore
-    private Authentication authentication = new AuthenticationDisabled();
+    private transient Authentication authentication = new AuthenticationDisabled();
+    @JsonIgnore
+    private transient String authPluginClassName;
+    @JsonIgnore
+    private transient Map<String, String> authParams;
+
     private long operationTimeoutMs = 30000;
     private long statsIntervalSeconds = 60;
 

diff --git a/...flink/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarProducer.java b/...flink/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarProducer.java
@@ -22,7 +22,6 @@
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
 import java.util.function.Function;
-import java.util.Map;
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.functions.RuntimeContext;

diff --git a/...flink/src/main/java/org/apache/flink/streaming/connectors/pulsar/PulsarSourceBuilder.java b/...flink/src/main/java/org/apache/flink/streaming/connectors/pulsar/PulsarSourceBuilder.java
@@ -27,6 +27,7 @@
 import org.apache.pulsar.client.api.AuthenticationFactory;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.impl.auth.AuthenticationToken;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
 
@@ -268,7 +269,7 @@ public PulsarSourceBuilder<T> pulsarAllConsumerConf(ConsumerConfigurationData co
     }
 
 
-    public SourceFunction<T> build() {
+    public SourceFunction<T> build() throws PulsarClientException{
         Preconditions.checkArgument(StringUtils.isNotBlank(this.clientConfigurationData.getServiceUrl()),
                 "a service url is required");
         Preconditions.checkArgument((this.consumerConfigurationData.getTopicNames() != null &&
@@ -277,9 +278,27 @@ public SourceFunction<T> build() {
                 "At least one topic or topics pattern is required");
         Preconditions.checkArgument(StringUtils.isNotBlank(this.consumerConfigurationData.getSubscriptionName()),
                 "a subscription name is required");
+
+        setTransientFields();
+
         return new PulsarConsumerSource<>(this);
     }
 
+    private void setTransientFields() throws PulsarClientException {
+        setAuth();
+    }
+
+    private void setAuth() throws PulsarClientException{
+        if (StringUtils.isBlank(this.clientConfigurationData.getAuthPluginClassName())
+                && this.clientConfigurationData.getAuthParams() == null || this.clientConfigurationData.getAuthParams().isEmpty())
+            return;
+
+        clientConfigurationData.setAuthentication(
+                AuthenticationFactory.create(
+                        this.clientConfigurationData.getAuthPluginClassName(),
+                        this.clientConfigurationData.getAuthParams()));
+    }
+
     /**
      * Creates a PulsarSourceBuilder.
      *

diff --git a/...k/src/test/java/org/apache/flink/streaming/connectors/pulsar/PulsarSourceBuilderTest.java b/...k/src/test/java/org/apache/flink/streaming/connectors/pulsar/PulsarSourceBuilderTest.java
@@ -22,6 +22,7 @@
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.SubscriptionInitialPosition;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
@@ -47,7 +48,7 @@ public void before() {
     }
 
     @Test
-    public void testBuild() {
+    public void testBuild() throws PulsarClientException {
         SourceFunction sourceFunction = pulsarSourceBuilder
                 .serviceUrl("testServiceUrl")
                 .topic("testTopic")
@@ -59,7 +60,7 @@ public void testBuild() {
 
 
     @Test
-    public void testBuildWithConfPojo() {
+    public void testBuildWithConfPojo() throws PulsarClientException {
         ClientConfigurationData clientConf = ClientConfigurationData.builder().serviceUrl("testServiceUrl").build();
         ConsumerConfigurationData consumerConf = ConsumerConfigurationData.builder()
                 .topicNames(new HashSet<>(Arrays.asList("testTopic")))
@@ -74,7 +75,7 @@ public void testBuildWithConfPojo() {
     }
 
     @Test(expectedExceptions = IllegalArgumentException.class)
-    public void testBuildWithoutSettingRequiredProperties() {
+    public void testBuildWithoutSettingRequiredProperties() throws PulsarClientException {
         pulsarSourceBuilder.build();
     }
 
@@ -158,7 +159,7 @@ public TypeInformation<T> getProducedType() {
     }
 
     @Test(expectedExceptions = IllegalArgumentException.class)
-    public void testServiceUrlNullWithConfPojo() {
+    public void testServiceUrlNullWithConfPojo() throws PulsarClientException {
         ClientConfigurationData clientConf = ClientConfigurationData.builder().serviceUrl(null).build();
         ConsumerConfigurationData consumerConf = ConsumerConfigurationData.builder()
                 .topicNames(new HashSet<String>(Arrays.asList("testServiceUrl")))
@@ -172,7 +173,7 @@ public void testServiceUrlNullWithConfPojo() {
     }
 
     @Test(expectedExceptions = IllegalArgumentException.class)
-    public void testServiceUrlWithBlankWithConfPojo() {
+    public void testServiceUrlWithBlankWithConfPojo() throws PulsarClientException {
         ClientConfigurationData clientConf = ClientConfigurationData.builder().serviceUrl(StringUtils.EMPTY).build();
         ConsumerConfigurationData consumerConf = ConsumerConfigurationData.builder()
                 .topicNames(new HashSet<String>(Arrays.asList("testTopic")))
@@ -186,7 +187,7 @@ public void testServiceUrlWithBlankWithConfPojo() {
     }
 
     @Test(expectedExceptions = IllegalArgumentException.class)
-    public void testTopicPatternWithNullWithConfPojo() {
+    public void testTopicPatternWithNullWithConfPojo() throws PulsarClientException {
         ClientConfigurationData clientConf = ClientConfigurationData.builder().serviceUrl("testServiceUrl").build();
         ConsumerConfigurationData consumerConf = ConsumerConfigurationData.builder()
                 .topicsPattern(null)
@@ -200,7 +201,7 @@ public void testTopicPatternWithNullWithConfPojo() {
     }
 
     @Test(expectedExceptions = IllegalArgumentException.class)
-    public void testSubscriptionNameWithNullWithConfPojo() {
+    public void testSubscriptionNameWithNullWithConfPojo() throws PulsarClientException {
         ClientConfigurationData clientConf = ClientConfigurationData.builder().serviceUrl("testServiceUrl").build();
         ConsumerConfigurationData consumerConf = ConsumerConfigurationData.builder()
                 .topicNames(new HashSet<String>(Arrays.asList("testTopic")))
@@ -214,7 +215,7 @@ public void testSubscriptionNameWithNullWithConfPojo() {
     }
 
     @Test(expectedExceptions = IllegalArgumentException.class)
-    public void testSubscriptionNameWithBlankWithConfPojo() {
+    public void testSubscriptionNameWithBlankWithConfPojo() throws PulsarClientException {
         pulsarSourceBuilder.topic(null);
         ClientConfigurationData clientConf = ClientConfigurationData.builder().serviceUrl("testServiceUrl").build();
         ConsumerConfigurationData consumerConf = ConsumerConfigurationData.builder()
@@ -229,7 +230,7 @@ public void testSubscriptionNameWithBlankWithConfPojo() {
     }
 
     @Test(expectedExceptions = IllegalArgumentException.class)
-    public void testSubscriptionInitialPositionWithConfPojo() {
+    public void testSubscriptionInitialPositionWithConfPojo() throws PulsarClientException {
         pulsarSourceBuilder.topic(null);
         ClientConfigurationData clientConf = ClientConfigurationData.builder().serviceUrl("testServiceUrl").build();
         ConsumerConfigurationData consumerConf = ConsumerConfigurationData.builder()