[HUDI-4616] Adding PulsarSource to DeltaStreamer to support ingesting from Apache Pulsar

hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/HoodieConversionUtils.scala

@@ -20,8 +20,21 @@ package org.apache.hudi
 
 import org.apache.hudi.common.config.TypedProperties
 
+import java.{util => ju}
+import scala.collection.JavaConverters
+
 object HoodieConversionUtils {
 
+  /**
+   * Converts Java's [[ju.Map]] into Scala's (immutable) [[Map]] (by default [[JavaConverters]] convert to
+   * a mutable one)
+   */
+  def mapAsScalaImmutableMap[K, V](map: ju.Map[K, V]): Map[K, V] = {
+    // NOTE: We have to use deprecated [[JavaConversions]] to stay compatible w/ Scala 2.11
+    import scala.collection.JavaConversions.mapAsScalaMap
+    map.toMap
+  }
+
   def toJavaOption[T](opt: Option[T]): org.apache.hudi.common.util.Option[T] =
     if (opt.isDefined) org.apache.hudi.common.util.Option.of(opt.get) else org.apache.hudi.common.util.Option.empty()
 

hudi-common/src/main/java/org/apache/hudi/common/util/ThreadUtils.java

@@ -0,0 +1,41 @@
+/*
+ * 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.hudi.common.util;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class ThreadUtils {
+
+  /**
+   * Fetches all active threads currently running in the JVM
+   */
+  public static List<Thread> collectActiveThreads() {
+    ThreadGroup threadGroup = Thread.currentThread().getThreadGroup();
+    while (threadGroup.getParent() != null) {
+      threadGroup = threadGroup.getParent();
+    }
+
+    Thread[] activeThreads = new Thread[threadGroup.activeCount()];
+    threadGroup.enumerate(activeThreads);
+
+    return Arrays.asList(activeThreads);
+  }
+
+}

hudi-utilities/pom.xml

@@ -171,6 +171,14 @@
       </exclusions>
     </dependency>
 
+    <!-- Pulsar Spark Connector -->
+    <dependency>
+      <groupId>io.streamnative.connectors</groupId>
+      <artifactId>pulsar-spark-connector_${scala.binary.version}</artifactId>
+      <version>${pulsar.spark.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
     <!-- Logging -->
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java

@@ -91,6 +91,7 @@
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.SparkSession;
 
+import java.io.Closeable;
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.ArrayList;
@@ -124,7 +125,7 @@
 /**
  * Sync's one batch of data to hoodie table.
  */
-public class DeltaSync implements Serializable {
+public class DeltaSync implements Serializable, Closeable {
 
   private static final long serialVersionUID = 1L;
   private static final Logger LOG = LogManager.getLogger(DeltaSync.class);
@@ -895,11 +896,15 @@ private void registerAvroSchemas(Schema sourceSchema, Schema targetSchema) {
    * Close all resources.
    */
   public void close() {
-    if (null != writeClient) {
+    if (writeClient != null) {
       writeClient.close();
       writeClient = null;
     }
 
+    if (formatAdapter != null) {
+      formatAdapter.close();
+    }
+
     LOG.info("Shutting down embedded timeline server");
     if (embeddedTimelineService.isPresent()) {
       embeddedTimelineService.get().stop();

hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/SourceFormatAdapter.java

@@ -21,6 +21,7 @@
 import org.apache.hudi.AvroConversionUtils;
 import org.apache.hudi.HoodieSparkUtils;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieIOException;
 import org.apache.hudi.utilities.UtilHelpers;
 import org.apache.hudi.utilities.schema.FilebasedSchemaProvider;
 import org.apache.hudi.utilities.schema.SchemaProvider;
@@ -38,13 +39,16 @@
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.types.StructType;
 
+import java.io.Closeable;
+import java.io.IOException;
+
 import static org.apache.hudi.utilities.schema.RowBasedSchemaProvider.HOODIE_RECORD_NAMESPACE;
 import static org.apache.hudi.utilities.schema.RowBasedSchemaProvider.HOODIE_RECORD_STRUCT_NAME;
 
 /**
  * Adapts data-format provided by the source to the data-format required by the client (DeltaStreamer).
  */
-public final class SourceFormatAdapter {
+public final class SourceFormatAdapter implements Closeable {
 
   private final Source source;
 
@@ -123,4 +127,15 @@ public InputBatch<Dataset<Row>> fetchNewDataInRowFormat(Option<String> lastCkptS
   public Source getSource() {
     return source;
   }
+
+  @Override
+  public void close() {
+    if (source instanceof Closeable) {
+      try {
+        ((Closeable) source).close();
+      } catch (IOException e) {
+        throw new HoodieIOException(String.format("Failed to shutdown the source (%s)", source.getClass().getName()), e);
+      }
+    }
+  }
 }