Revert "Enable pulsar function to send message to external pulsar clu… (

apache#10843) This reverts commit 66231e3. ### Motivation Reverting since the external cluster configuration was not not needed as there are better ways to interact with external pulsar clusters. A user can simply instantiate a Pulsar client, producer, or consumer in a function to interact with another pulsar cluster. This feature is also a security risk as it does not security handle authentication data and it will be written as plain text in the internal function's metadata topic.
mfleming · Jun 7, 2021 · ed10d82 · ed10d82
1 parent 02d5eb3
commit ed10d82
Show file tree

Hide file tree

Showing 52 changed files with 220 additions and 342 deletions.
diff --git a/...ient-admin-api/src/main/java/org/apache/pulsar/common/functions/AuthenticationConfig.java b/...ient-admin-api/src/main/java/org/apache/pulsar/common/functions/AuthenticationConfig.java
@@ -18,14 +18,18 @@
  */
 package org.apache.pulsar.common.functions;
 
+import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
+import lombok.NoArgsConstructor;
 
 /**
  * Configuration to aggregate various authentication params.
  */
 @Data
 @Builder
+@AllArgsConstructor
+@NoArgsConstructor
 public class AuthenticationConfig {
     private String clientAuthenticationPlugin;
     private String clientAuthenticationParameters;

diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/functions/FunctionConfig.java
@@ -95,7 +95,6 @@ public enum Runtime {
     private String batchBuilder;
     private Boolean forwardSourceMessageProperty;
     private Map<String, Object> userConfig;
-    private Map<String, ExternalPulsarConfig> externalPulsars;
     // This is a map of secretName(aka how the secret is going to be
     // accessed in the function via context) to an object that
     // encapsulates how the secret is fetched by the underlying

diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
@@ -52,7 +52,6 @@
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.SubscriptionInitialPosition;
 import org.apache.pulsar.common.functions.ConsumerConfig;
-import org.apache.pulsar.common.functions.ExternalPulsarConfig;
 import org.apache.pulsar.common.functions.FunctionConfig;
 import org.apache.pulsar.common.functions.ProducerConfig;
 import org.apache.pulsar.common.functions.Resources;
@@ -322,8 +321,6 @@ abstract class FunctionDetailsCommand extends BaseCommand {
         protected String customRuntimeOptions;
         @Parameter(names = "--dead-letter-topic", description = "The topic where messages that are not processed successfully are sent to")
         protected String deadLetterTopic;
-        @Parameter(names = "--external-pulsars", description = "The map of external pulsar cluster name to its configuration (as a JSON string)")
-        protected String externalPulsars;
         protected FunctionConfig functionConfig;
         protected String userCodeFile;
 
@@ -402,11 +399,6 @@ void processArguments() throws Exception {
             if (null != output) {
                 functionConfig.setOutput(output);
             }
-            if (null != externalPulsars) {
-                Type type = new TypeToken<Map<String, ExternalPulsarConfig>>() {
-                }.getType();
-                functionConfig.setExternalPulsars(new Gson().fromJson(externalPulsars, type));
-            }
             if (null != producerConfig) {
                 Type type = new TypeToken<ProducerConfig>() {}.getType();
                 functionConfig.setProducerConfig(new Gson().fromJson(producerConfig, type));

diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java b/pulsar-common/src/main/java/org/apache/pulsar/common/naming/TopicName.java
@@ -78,7 +78,7 @@ public static TopicName get(String domain, String tenant, String namespace, Stri
     }
 
     public static TopicName get(String domain, String tenant, String cluster, String namespace,
-            String topic) {
+                                String topic) {
         String name = domain + "://" + tenant + '/' + cluster + '/' + namespace + '/' + topic;
         return TopicName.get(name);
     }

diff --git a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Context.java
@@ -271,14 +271,6 @@ public interface Context {
      */
     PulsarAdmin getPulsarAdmin();
 
-    /**
-     * Get the pulsar admin client by cluster name.
-     *
-     * @param clusterName The name of the cluster name for pulsar admin client
-     * @return The instance of pulsar admin client
-     */
-    PulsarAdmin getPulsarAdmin(String clusterName);
-
     /**
      * Record a user defined metric.
      *
@@ -320,18 +312,6 @@ public interface Context {
      */
     <O> TypedMessageBuilder<O> newOutputMessage(String topicName, Schema<O> schema) throws PulsarClientException;
 
-    /**
-     * New output message using schema for serializing to the topic in the cluster
-     *
-     * @param clusterName the name of the cluster for topic
-     * @param topicName The name of the topic for output message
-     * @param schema provide a way to convert between serialized data and domain objects
-     * @param <O>
-     * @return the message builder instance
-     * @throws PulsarClientException
-     */
-    <O> TypedMessageBuilder<O> newOutputMessage(String clusterName, String topicName, Schema<O> schema) throws PulsarClientException;
-
     /**
      * Create a ConsumerBuilder with the schema.
      *
@@ -341,4 +321,4 @@ public interface Context {
      * @throws PulsarClientException
      */
     <O> ConsumerBuilder<O> newConsumerBuilder(Schema<O> schema) throws PulsarClientException;
-}
+}
diff --git a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java
@@ -18,12 +18,11 @@
  */
 package org.apache.pulsar.functions.api;
 
-import org.apache.pulsar.client.api.Message;
-import org.apache.pulsar.client.api.Schema;
-
 import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.common.classification.InterfaceAudience;
 import org.apache.pulsar.common.classification.InterfaceStability;
 

diff --git a/...ommon/functions/ExternalPulsarConfig.java → ...ctions/instance/AuthenticationConfig.java b/...ommon/functions/ExternalPulsarConfig.java → ...ctions/instance/AuthenticationConfig.java
@@ -16,27 +16,21 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+package org.apache.pulsar.functions.instance;
 
-package org.apache.pulsar.common.functions;
-
-import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.NoArgsConstructor;
 
 /**
- * Configuration of extra pulsar clusters to sent output message.
+ * Configuration to aggregate various authentication params.
  */
 @Data
 @Builder
-@NoArgsConstructor
-@AllArgsConstructor
-@EqualsAndHashCode
-public class ExternalPulsarConfig {
-    private String name;
-    private String serviceURL;
-    private String webServiceURL;
-    private AuthenticationConfig authConfig;
-    private ProducerConfig producerConfig;
+public class AuthenticationConfig {
+    private String clientAuthenticationPlugin;
+    private String clientAuthenticationParameters;
+    private String tlsTrustCertsFilePath;
+    private boolean useTls;
+    private boolean tlsAllowInsecureConnection;
+    private boolean tlsHostnameVerificationEnable;
 }