You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/05/26 08:08:00 UTC

[GitHub] [pulsar-adapters] atezs82 opened a new pull request #14: [pulsar-spark] added option for configuring Pulsar client

atezs82 opened a new pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14


   <!--
   ### Contribution Checklist
     
     - Name the pull request in the form "[Issue XYZ][component] Title of the pull request", where *XYZ* should be replaced by the actual issue number.
       Skip *Issue XYZ* if there is no associated github issue for this pull request.
       Skip *component* if you are unsure about which is the best component. E.g. `[docs] Fix typo in produce method`.
   
     - Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
     
     - Each pull request should address only one issue, not mix up code from multiple issues.
     
     - Each commit in the pull request has a meaningful commit message
   
     - Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
   
   **(The sections below can be removed for hotfixes of typos)**
   -->
   ### Motivation
   
   It is currently not possible to configure the Pulsar Client when creating a `SparkStreamingPulsarReceiver` (currently only the service URL, the authentication plugin and the consumer configuration can be set). This renders the receiver unusable when eg. the Pulsar cluster is configured to use TLS, and custom TLS certificates need to be set.
   
   ### Modifications
   
   - Added an optional Map to configure the client to the constructor of the receiver. Existing constructors still remain usable, but they do not set this parameter. When the parameter is an empty map, `loadConf` on the client builder does not get called. Existing constructors set this parameter to an empty map to maintain backward compatibility.
   - If specified, (since `ClientBuilder` works like this) the configuration overrides other client configuration.
   - Added a simple integration test around the functionality (it barely checks whether `serviceUrl` can be overriden by this feature).
   - Fixed some unit tests around `SparkStreamingPulsarReceiver` (since they seemed to use an older interface). This issue was similar to https://github.com/apache/pulsar-adapters/issues/12 (but not exactly the same).
   - If the modifications look good, I also plan to modify the documentation accordingly.
   
   ### Verifying this change
   
   - [ ] Make sure that the change passes the CI checks.
   
   This change added tests and can be verified as follows:
   - Added a simple integration test around the functionality (it barely checks whether `serviceUrl` can be overriden by this feature).
   - Existing integration tests for the receiver were also fixed and observed to be working during local run.
   
   ### Does this pull request potentially affect one of the following parts:
   
   *If `yes` was chosen, please highlight the changes*
   
     - Dependencies (does it add or upgrade a dependency): no
     - The public API: yes (ability to set client configuration when creating receiver)
     - The schema: don't know
     - The default values of configurations: no
     - The wire protocol: no
     - The rest endpoints: no
     - The admin cli options: no
     - Anything that affects deployment: don't know
   
   ### Documentation
   
     - Does this pull request introduce a new feature? yes
     - If yes, how is the feature documented? will do docs and Javadocs once the feature looks good in this PR
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] atezs82 commented on pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
atezs82 commented on pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#issuecomment-864457360


   Thanks for the comments, removed all non-Spark related changes from PR. Due to some issues, I still do not have tests working. I'm not sure whether this is a local or general problem but I get `package org.apache.pulsar.tests.integration.suites does not exist` during test run, will need to take a look into it a bit later.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] sijie commented on a change in pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
sijie commented on a change in pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#discussion_r657609793



##########
File path: pulsar-spark/src/main/java/org/apache/pulsar/spark/SparkStreamingPulsarReceiver.java
##########
@@ -43,34 +46,52 @@
     private static final Logger LOG = LoggerFactory.getLogger(SparkStreamingPulsarReceiver.class);
 
     private String serviceUrl;
-    private ConsumerConfigurationData<byte[]> conf;
+    private Map<String,Object> clientConfig;
+    private ConsumerConfigurationData<byte[]> consumerConfig;
     private Authentication authentication;
     private PulsarClient pulsarClient;
     private Consumer<byte[]> consumer;
 
     public SparkStreamingPulsarReceiver(
         String serviceUrl,
-        ConsumerConfigurationData<byte[]> conf,
+        ConsumerConfigurationData<byte[]> consumerConfig,
         Authentication authentication) {
-        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, conf, authentication);
+        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, new HashMap<>(), consumerConfig, authentication);
+    }
+
+    public SparkStreamingPulsarReceiver(
+            String serviceUrl,
+            Map<String,Object> clientConfig,
+            ConsumerConfigurationData<byte[]> consumerConfig,
+            Authentication authentication) {
+        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, clientConfig, consumerConfig, authentication);
+    }
+
+    public SparkStreamingPulsarReceiver(StorageLevel storageLevel,

Review comment:
       We are introducing a new parameter `storageLevel`. But it was not used.

##########
File path: pulsar-spark/src/main/java/org/apache/pulsar/spark/SparkStreamingPulsarReceiver.java
##########
@@ -43,34 +46,52 @@
     private static final Logger LOG = LoggerFactory.getLogger(SparkStreamingPulsarReceiver.class);
 
     private String serviceUrl;
-    private ConsumerConfigurationData<byte[]> conf;
+    private Map<String,Object> clientConfig;
+    private ConsumerConfigurationData<byte[]> consumerConfig;
     private Authentication authentication;
     private PulsarClient pulsarClient;
     private Consumer<byte[]> consumer;
 
     public SparkStreamingPulsarReceiver(
         String serviceUrl,
-        ConsumerConfigurationData<byte[]> conf,
+        ConsumerConfigurationData<byte[]> consumerConfig,
         Authentication authentication) {
-        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, conf, authentication);
+        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, new HashMap<>(), consumerConfig, authentication);
+    }
+
+    public SparkStreamingPulsarReceiver(

Review comment:
       Why we are going to introduce so many new builders?




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] eolivelli commented on pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#issuecomment-864455395


   We are already working on fixing integration tests
   
   Ias you are touching only the spark adapter then you can ignore the Kafka stuff
   
   Thanks


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] atezs82 commented on pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
atezs82 commented on pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#issuecomment-864600449


   @lhotari Thanks for the workaround, I have managed to run all tests successfully locally for the PR.
   
   @eolivelli Sorry for the mess, I have corrected the import statement.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] lhotari edited a comment on pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
lhotari edited a comment on pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#issuecomment-864554978


   > @eolivelli Rebased on top of master, but I might have some additional work to do, since tests are failing still locally for `Apache Pulsar :: Tests :: Pulsar Kafka Compat Client Tests`. Please let me know what you think about this change, I can work further on fixing local issues in the meantime.
   
   That is due to the fact the the integration test library hasn't been published to maven central. The workaround is to compile and install it locally. Here are the commands used in CI:
   https://github.com/apache/pulsar-adapters/blob/9801c43898f1b31c6335382c5ea16eac200b603d/.github/workflows/integration-test.yaml#L74-L90
   
   The commands also cover building `apachepulsar/pulsar-test-latest-version:latest` used by integration tests since the one in docker hub isn't up-to-date.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] atezs82 commented on a change in pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
atezs82 commented on a change in pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#discussion_r654834504



##########
File path: pulsar-spark/src/main/java/org/apache/pulsar/spark/SparkStreamingPulsarReceiver.java
##########
@@ -22,13 +22,12 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
 
-import org.apache.pulsar.client.api.Authentication;
-import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.MessageListener;
-import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.*;

Review comment:
       Corrected.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] atezs82 commented on a change in pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
atezs82 commented on a change in pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#discussion_r654834560



##########
File path: pulsar-flink/src/test/java/org/apache/flink/streaming/connectors/pulsar/PulsarConsumerSourceTests.java
##########
@@ -57,6 +57,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;

Review comment:
       Removed this as well.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] atezs82 commented on pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
atezs82 commented on pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#issuecomment-864454151


   @eolivelli Rebased on top of master, but I might have some additional work to do, since tests are failing still locally for `Apache Pulsar :: Tests :: Pulsar Kafka Compat Client Tests`. Please let me know what you think about this change, I can work further on fixing local issues in the meantime.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] eolivelli commented on pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#issuecomment-866301922


   @codelipenghui @sijie PTAL


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] lhotari commented on pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#issuecomment-864554978


   > @eolivelli Rebased on top of master, but I might have some additional work to do, since tests are failing still locally for `Apache Pulsar :: Tests :: Pulsar Kafka Compat Client Tests`. Please let me know what you think about this change, I can work further on fixing local issues in the meantime.
   
   That is due to the fact the the integration test library hasn't been published to maven central. The workaround is to compile and install it locally. Here are the commands used in CI:
   https://github.com/apache/pulsar-adapters/blob/9801c43898f1b31c6335382c5ea16eac200b603d/.github/workflows/integration-test.yaml#L74-L90
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] atezs82 commented on a change in pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
atezs82 commented on a change in pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#discussion_r697829512



##########
File path: pulsar-spark/src/main/java/org/apache/pulsar/spark/SparkStreamingPulsarReceiver.java
##########
@@ -43,34 +46,52 @@
     private static final Logger LOG = LoggerFactory.getLogger(SparkStreamingPulsarReceiver.class);
 
     private String serviceUrl;
-    private ConsumerConfigurationData<byte[]> conf;
+    private Map<String,Object> clientConfig;
+    private ConsumerConfigurationData<byte[]> consumerConfig;
     private Authentication authentication;
     private PulsarClient pulsarClient;
     private Consumer<byte[]> consumer;
 
     public SparkStreamingPulsarReceiver(
         String serviceUrl,
-        ConsumerConfigurationData<byte[]> conf,
+        ConsumerConfigurationData<byte[]> consumerConfig,
         Authentication authentication) {
-        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, conf, authentication);
+        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, new HashMap<>(), consumerConfig, authentication);
+    }
+
+    public SparkStreamingPulsarReceiver(

Review comment:
       @sijie What do you think about this proposal? Shall I change the code like this?




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] eolivelli commented on a change in pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#discussion_r654908141



##########
File path: pulsar-spark/src/main/java/org/apache/pulsar/spark/SparkStreamingPulsarReceiver.java
##########
@@ -22,13 +22,12 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
 
-import org.apache.pulsar.client.api.Authentication;
-import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.MessageListener;
-import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.*;

Review comment:
       did you push ?
   I still see a star import here on GitHub




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] atezs82 commented on a change in pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
atezs82 commented on a change in pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#discussion_r657762984



##########
File path: pulsar-spark/src/main/java/org/apache/pulsar/spark/SparkStreamingPulsarReceiver.java
##########
@@ -43,34 +46,52 @@
     private static final Logger LOG = LoggerFactory.getLogger(SparkStreamingPulsarReceiver.class);
 
     private String serviceUrl;
-    private ConsumerConfigurationData<byte[]> conf;
+    private Map<String,Object> clientConfig;
+    private ConsumerConfigurationData<byte[]> consumerConfig;
     private Authentication authentication;
     private PulsarClient pulsarClient;
     private Consumer<byte[]> consumer;
 
     public SparkStreamingPulsarReceiver(
         String serviceUrl,
-        ConsumerConfigurationData<byte[]> conf,
+        ConsumerConfigurationData<byte[]> consumerConfig,
         Authentication authentication) {
-        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, conf, authentication);
+        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, new HashMap<>(), consumerConfig, authentication);
+    }
+
+    public SparkStreamingPulsarReceiver(
+            String serviceUrl,
+            Map<String,Object> clientConfig,
+            ConsumerConfigurationData<byte[]> consumerConfig,
+            Authentication authentication) {
+        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, clientConfig, consumerConfig, authentication);
+    }
+
+    public SparkStreamingPulsarReceiver(StorageLevel storageLevel,

Review comment:
       Thanks for the finding, corrected this.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] atezs82 commented on a change in pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
atezs82 commented on a change in pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#discussion_r657772386



##########
File path: pulsar-spark/src/main/java/org/apache/pulsar/spark/SparkStreamingPulsarReceiver.java
##########
@@ -43,34 +46,52 @@
     private static final Logger LOG = LoggerFactory.getLogger(SparkStreamingPulsarReceiver.class);
 
     private String serviceUrl;
-    private ConsumerConfigurationData<byte[]> conf;
+    private Map<String,Object> clientConfig;
+    private ConsumerConfigurationData<byte[]> consumerConfig;
     private Authentication authentication;
     private PulsarClient pulsarClient;
     private Consumer<byte[]> consumer;
 
     public SparkStreamingPulsarReceiver(
         String serviceUrl,
-        ConsumerConfigurationData<byte[]> conf,
+        ConsumerConfigurationData<byte[]> consumerConfig,
         Authentication authentication) {
-        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, conf, authentication);
+        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, new HashMap<>(), consumerConfig, authentication);
+    }
+
+    public SparkStreamingPulsarReceiver(

Review comment:
       I would be happy to remove eg. the one having a default storage level: 
   ```
   public SparkStreamingPulsarReceiver(
               String serviceUrl,
               Map<String,Object> clientConfig,
               ConsumerConfigurationData<byte[]> consumerConfig,
               Authentication authentication) {
           this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, clientConfig, consumerConfig, authentication);
       }
   ```
   In that way we might have broader functionality (we just enforce the user to set the storage level whenever client configuration is provided).




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] atezs82 commented on a change in pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
atezs82 commented on a change in pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#discussion_r657762984



##########
File path: pulsar-spark/src/main/java/org/apache/pulsar/spark/SparkStreamingPulsarReceiver.java
##########
@@ -43,34 +46,52 @@
     private static final Logger LOG = LoggerFactory.getLogger(SparkStreamingPulsarReceiver.class);
 
     private String serviceUrl;
-    private ConsumerConfigurationData<byte[]> conf;
+    private Map<String,Object> clientConfig;
+    private ConsumerConfigurationData<byte[]> consumerConfig;
     private Authentication authentication;
     private PulsarClient pulsarClient;
     private Consumer<byte[]> consumer;
 
     public SparkStreamingPulsarReceiver(
         String serviceUrl,
-        ConsumerConfigurationData<byte[]> conf,
+        ConsumerConfigurationData<byte[]> consumerConfig,
         Authentication authentication) {
-        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, conf, authentication);
+        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, new HashMap<>(), consumerConfig, authentication);
+    }
+
+    public SparkStreamingPulsarReceiver(
+            String serviceUrl,
+            Map<String,Object> clientConfig,
+            ConsumerConfigurationData<byte[]> consumerConfig,
+            Authentication authentication) {
+        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, clientConfig, consumerConfig, authentication);
+    }
+
+    public SparkStreamingPulsarReceiver(StorageLevel storageLevel,

Review comment:
       Thanks for the finding, corrected this in the upcoming patchset.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] eolivelli commented on pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
eolivelli commented on pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#issuecomment-863849216


   @atezs82 can you please rebase this patch?
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] eolivelli commented on a change in pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
eolivelli commented on a change in pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#discussion_r654833418



##########
File path: pulsar-spark/src/main/java/org/apache/pulsar/spark/SparkStreamingPulsarReceiver.java
##########
@@ -22,13 +22,12 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
 
-import org.apache.pulsar.client.api.Authentication;
-import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.MessageListener;
-import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.*;

Review comment:
       Please do not use star import.
   

##########
File path: tests/pulsar-kafka-compat-client-test/src/test/java/org/apache/pulsar/tests/integration/compat/kafka/PulsarKafkaProducerThreadSafeTest.java
##########
@@ -37,8 +36,8 @@ private String getPlainTextServiceUrl() {
         return container.getPlainTextServiceUrl();
     }
 
-    @Override
-    public void setUpCluster() throws Exception {
+    @BeforeTest
+    private void setupThreadSafeTest() {

Review comment:
       This change is unrelated can you please revert?
   
   Cc @dlg99 @lhotari 

##########
File path: pulsar-flink/src/test/java/org/apache/flink/streaming/connectors/pulsar/PulsarConsumerSourceTests.java
##########
@@ -57,6 +57,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;

Review comment:
       This change probably not needed 




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] atezs82 commented on a change in pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
atezs82 commented on a change in pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#discussion_r657769442



##########
File path: pulsar-spark/src/main/java/org/apache/pulsar/spark/SparkStreamingPulsarReceiver.java
##########
@@ -43,34 +46,52 @@
     private static final Logger LOG = LoggerFactory.getLogger(SparkStreamingPulsarReceiver.class);
 
     private String serviceUrl;
-    private ConsumerConfigurationData<byte[]> conf;
+    private Map<String,Object> clientConfig;
+    private ConsumerConfigurationData<byte[]> consumerConfig;
     private Authentication authentication;
     private PulsarClient pulsarClient;
     private Consumer<byte[]> consumer;
 
     public SparkStreamingPulsarReceiver(
         String serviceUrl,
-        ConsumerConfigurationData<byte[]> conf,
+        ConsumerConfigurationData<byte[]> consumerConfig,
         Authentication authentication) {
-        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, conf, authentication);
+        this(StorageLevel.MEMORY_AND_DISK_2(), serviceUrl, new HashMap<>(), consumerConfig, authentication);
+    }
+
+    public SparkStreamingPulsarReceiver(

Review comment:
       The idea behind adding 2 new constructors to the 2 existing ones is I wanted to allow users of the interface to be able to configure Pulsar clients regardless whether they want to specify a storage level or leave it to default. Also, I wanted to keep previous constructors as well to keep backward compatibility.
   
   Please let me know what do you think about this.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] eolivelli edited a comment on pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
eolivelli edited a comment on pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#issuecomment-864455395


   We are already working on fixing integration tests
   
   as you are touching only the spark adapter then you can ignore the Kafka stuff
   
   Thanks


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



[GitHub] [pulsar-adapters] atezs82 commented on a change in pull request #14: [pulsar-spark] added option for configuring Pulsar client

Posted by GitBox <gi...@apache.org>.
atezs82 commented on a change in pull request #14:
URL: https://github.com/apache/pulsar-adapters/pull/14#discussion_r654834620



##########
File path: tests/pulsar-kafka-compat-client-test/src/test/java/org/apache/pulsar/tests/integration/compat/kafka/PulsarKafkaProducerThreadSafeTest.java
##########
@@ -37,8 +36,8 @@ private String getPlainTextServiceUrl() {
         return container.getPlainTextServiceUrl();
     }
 
-    @Override
-    public void setUpCluster() throws Exception {
+    @BeforeTest
+    private void setupThreadSafeTest() {

Review comment:
       Removed this from here.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org