Extend PulsarIO to support Batch Sources. This pr implements PIP-65

pulsar-common/src/main/java/org/apache/pulsar/common/io/BatchSourceConfig.java

@@ -0,0 +1,50 @@
+/**
+ * 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.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.common.io;
+
+import java.util.Map;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import lombok.ToString;
+
+/**
+ * Pulsar Batch Source configuration.
+ */
+@Getter
+@Setter
+@Data
+@EqualsAndHashCode
+@ToString
+@Builder(toBuilder = true)
+@NoArgsConstructor
+@AllArgsConstructor
+public class BatchSourceConfig {
+    public static final String BATCHSOURCE_CONFIG_KEY = "__BATCHSOURCECONFIGS__";
+    public static final String BATCHSOURCE_CLASSNAME_KEY = "__BATCHSOURCECLASSNAME__";
+
+    // The class used for triggering the discovery process
+    private String discoveryTriggererClassName;
+    // The config needed for the discovery Triggerer init
+    private Map<String, Object> discoveryTriggererConfig;
+}

pulsar-common/src/main/java/org/apache/pulsar/common/io/SourceConfig.java

@@ -63,4 +63,7 @@ public class SourceConfig {
     // to change behavior at runtime. Currently, this primarily used by the KubernetesManifestCustomizer
     // interface
     private String customRuntimeOptions;
+
+    // If this is a BatchSource, its batch related configs are stored here
+    private BatchSourceConfig batchSourceConfig;
 }

pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SourceConfigUtils.java

@@ -29,8 +29,10 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.common.functions.Resources;
+import org.apache.pulsar.common.io.BatchSourceConfig;
 import org.apache.pulsar.common.io.ConnectorDefinition;
 import org.apache.pulsar.common.io.SourceConfig;
+import org.apache.pulsar.common.naming.TopicDomain;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.nar.NarClassLoader;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
@@ -89,6 +91,7 @@ public static FunctionDetails convert(SourceConfig sourceConfig, ExtractedSource
             functionDetailsBuilder.setProcessingGuarantees(
                     convertProcessingGuarantee(sourceConfig.getProcessingGuarantees()));
         }
+
         // set source spec
         Function.SourceSpec.Builder sourceSpecBuilder = Function.SourceSpec.newBuilder();
         if (sourceDetails.getSourceClassName() != null) {
@@ -100,10 +103,21 @@ public static FunctionDetails convert(SourceConfig sourceConfig, ExtractedSource
             sourceSpecBuilder.setBuiltin(builtin);
         }
 
+        Map<String, Object> configs = new HashMap<>();
         if (sourceConfig.getConfigs() != null) {
-            sourceSpecBuilder.setConfigs(new Gson().toJson(sourceConfig.getConfigs()));
+            configs.putAll(sourceConfig.getConfigs());
+        }
+
+        // Batch source handling
+        if (sourceConfig.getBatchSourceConfig() != null) {
+            configs.put(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY, new Gson().toJson(sourceConfig.getBatchSourceConfig()));
+            configs.put(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY, sourceSpecBuilder.getClassName());
+            sourceSpecBuilder.setClassName("org.apache.pulsar.io.batch.BatchSourceExecutor");
         }
 
+        sourceSpecBuilder.setConfigs(new Gson().toJson(configs));
+
+
         if (sourceConfig.getSecrets() != null && !sourceConfig.getSecrets().isEmpty()) {
             functionDetailsBuilder.setSecretsMap(new Gson().toJson(sourceConfig.getSecrets()));
         }
@@ -167,16 +181,22 @@ public static SourceConfig convertFromDetails(FunctionDetails functionDetails) {
         if (!StringUtils.isEmpty(sourceSpec.getBuiltin())) {
             sourceConfig.setArchive("builtin://" + sourceSpec.getBuiltin());
         }
-        if (!StringUtils.isEmpty(sourceSpec.getConfigs())) {
-            TypeReference<HashMap<String,Object>> typeRef
-                    = new TypeReference<HashMap<String,Object>>() {};
-            Map<String, Object> configMap;
-            try {
-                configMap = ObjectMapperFactory.getThreadLocal().readValue(sourceSpec.getConfigs(), typeRef);
-            } catch (IOException e) {
-                log.error("Failed to read configs for source {}", FunctionCommon.getFullyQualifiedName(functionDetails), e);
-                throw new RuntimeException(e);
+        Map<String, Object> configMap = extractSourceConfig(sourceSpec, FunctionCommon.getFullyQualifiedName(functionDetails));
+        if (configMap != null) {
+            BatchSourceConfig batchSourceConfig = extractBatchSourceConfig(configMap);
+            if (batchSourceConfig != null) {
+                sourceConfig.setBatchSourceConfig(batchSourceConfig);
+                if (configMap.containsKey(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY)) {
+                    if (!StringUtils.isEmpty((String)configMap.get(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY))) {
+                        sourceConfig.setClassName((String)configMap.get(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY));
+                    } else {
+                        sourceConfig.setClassName(null);
+                    }
+                }
             }
+
+            configMap.remove(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY);
+            configMap.remove(BatchSourceConfig.BATCHSOURCE_CLASSNAME_KEY);
             sourceConfig.setConfigs(configMap);
         }
         if (!isEmpty(functionDetails.getSecretsMap())) {
@@ -346,6 +366,10 @@ public static ExtractedSourceDetails validate(SourceConfig sourceConfig, Path ar
             ValidatorUtils.validateSchema(sourceConfig.getSchemaType(), typeArg, classLoader, false);
         }
 
+        if (sourceConfig.getBatchSourceConfig() != null) {
+            validateBatchSourceConfig(sourceConfig.getBatchSourceConfig());
+        }
+
         return new ExtractedSourceDetails(sourceClassName, typeArg.getName());
     }
 
@@ -396,9 +420,86 @@ public static SourceConfig validateUpdate(SourceConfig existingConfig, SourceCon
         if (!StringUtils.isEmpty(newConfig.getCustomRuntimeOptions())) {
             mergedConfig.setCustomRuntimeOptions(newConfig.getCustomRuntimeOptions());
         }
+        if (isBatchSource(existingConfig) != isBatchSource(newConfig)) {
+            throw new IllegalArgumentException("Sources cannot be update between regular sources and batchsource");
+        }
+        if (newConfig.getBatchSourceConfig() != null) {
+            validateBatchSourceConfigUpdate(existingConfig.getBatchSourceConfig(), newConfig.getBatchSourceConfig());
+            mergedConfig.setBatchSourceConfig(newConfig.getBatchSourceConfig());
+        }
         return mergedConfig;
     }
 
+    public static void validateBatchSourceConfig(BatchSourceConfig batchSourceConfig) throws IllegalArgumentException {
+        if (isEmpty(batchSourceConfig.getDiscoveryTriggererClassName())) {
+            log.error("BatchSourceConfig does not specify Discovery Trigger ClassName");
+            throw new IllegalArgumentException("BatchSourceConfig does not specify Discovery Trigger ClassName");
+        }
+    }
+
+    public static Map<String, Object> extractSourceConfig(Function.SourceSpec sourceSpec, String fqfn) {
+        if (!StringUtils.isEmpty(sourceSpec.getConfigs())) {
+            TypeReference<HashMap<String, Object>> typeRef
+                    = new TypeReference<HashMap<String, Object>>() {
+            };
+            try {
+                return ObjectMapperFactory.getThreadLocal().readValue(sourceSpec.getConfigs(), typeRef);
+            } catch (IOException e) {
+                log.error("Failed to read configs for source {}", fqfn, e);
+                throw new RuntimeException(e);
+            }
+        } else {
+            return null;
+        }
+    }
+
+    public static BatchSourceConfig extractBatchSourceConfig(Map<String, Object> configMap) {
+        if (configMap.containsKey(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY)) {
+            String batchSourceConfigJson = (String) configMap.get(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY);
+            return new Gson().fromJson(batchSourceConfigJson, BatchSourceConfig.class);
+        } else {
+            return null;
+        }
+    }
+
+    public static Map<String, String> computeBatchSourceIntermediateTopicSubscriptions(Function.FunctionDetails details,
+                                                                                       String fqfn) {
+        Map<String, Object> configMap = extractSourceConfig(details.getSource(), fqfn);
+        if (configMap != null) {
+            BatchSourceConfig batchSourceConfig = extractBatchSourceConfig(configMap);
+            String intermediateTopicName = computeBatchSourceIntermediateTopicName(details.getTenant(),
+                    details.getNamespace(), details.getName()).toString();
+            if (batchSourceConfig != null) {
+                Map<String, String> subscriptionMap = new HashMap<>();
+                subscriptionMap.put(intermediateTopicName,
+                            computeBatchSourceInstanceSubscriptionName(details.getTenant(),
+                                    details.getNamespace(), details.getName()));
+                return subscriptionMap;
+            }
+        }
+        return null;
+    }
+
+    public static String computeBatchSourceInstanceSubscriptionName(String tenant, String namespace,
+                                                                    String sourceName) {
+        return "BatchSourceExecutor-" + tenant + "/" + namespace + "/" + sourceName;
+    }
+
+    public static TopicName computeBatchSourceIntermediateTopicName(String tenant, String namespace,
+                                                                    String sourceName) {
+        return TopicName.get(TopicDomain.persistent.name(), tenant, namespace, sourceName + "-intermediate");
+    }
+
+    public static boolean isBatchSource(SourceConfig sourceConfig) {
+        return sourceConfig.getBatchSourceConfig() != null;
+    }
+
+    public static void validateBatchSourceConfigUpdate(BatchSourceConfig existingConfig, BatchSourceConfig newConfig) {
+        if (!existingConfig.getDiscoveryTriggererClassName().equals(newConfig.getDiscoveryTriggererClassName())) {
+            throw new IllegalArgumentException("DiscoverTriggerer class cannot be updated for batchsources");
+        }
+    }
+
     public static void validateConnectorConfig(SourceConfig sourceConfig, ClassLoader classLoader) {
         try {
             ConnectorDefinition defn = ConnectorUtils.getConnectorDefinition(classLoader);
@@ -417,5 +518,4 @@ public static void validateConnectorConfig(SourceConfig sourceConfig, ClassLoade
             throw new IllegalArgumentException("Could not validate source config: " + e.getMessage());
         }
     }
-
 }