[HUDI-2937] Introduce a pulsar implementation of hoodie write commit callback

Author: XuQianJin-Stars

hudi-utilities/src/main/java/org/apache/hudi/utilities/callback/pulsar/HoodieWriteCommitPulsarCallback.java

@@ -78,7 +78,6 @@ public HoodieWriteCommitPulsarCallback(HoodieWriteConfig config) throws PulsarCl
     this.topic = config.getString(TOPIC);
     this.client = createClient(config);
     this.producer = createProducer(config);
-    validatePulsarConfig();
   }
 
   @Override
@@ -121,8 +120,9 @@ public Producer<String> createProducer(HoodieConfig hoodieConfig) throws PulsarC
         .create();
   }
 
-  public static PulsarClient createClient(HoodieConfig hoodieConfig) throws PulsarClientException {
-    String serviceUrl = hoodieConfig.getString(BROKER_SERVICE_URL);
+  public PulsarClient createClient(HoodieConfig hoodieConfig) throws PulsarClientException {
+    validatePulsarConfig();
+
     Duration operationTimeout =
         parseDuration(hoodieConfig.getString(OPERATION_TIMEOUT));
     Duration connectionTimeout =

hudi-utilities/src/main/java/org/apache/hudi/utilities/callback/pulsar/HoodieWriteCommitPulsarCallbackConfig.java

@@ -100,4 +100,19 @@ public class HoodieWriteCommitPulsarCallbackConfig extends HoodieConfig {
       .sinceVersion("0.11.0")
       .withDocumentation("Duration of keeping alive interval for each "
           + "client broker connection.");
+
+  /**
+   * Set default value for {@link HoodieWriteCommitPulsarCallbackConfig} if needed.
+   */
+  public static void setCallbackPulsarConfigIfNeeded(HoodieConfig config) {
+    config.setDefaultValue(PRODUCER_ROUTE_MODE);
+    config.setDefaultValue(OPERATION_TIMEOUT);
+    config.setDefaultValue(CONNECTION_TIMEOUT);
+    config.setDefaultValue(REQUEST_TIMEOUT);
+    config.setDefaultValue(KEEPALIVE_INTERVAL);
+    config.setDefaultValue(PRODUCER_SEND_TIMEOUT);
+    config.setDefaultValue(PRODUCER_PENDING_QUEUE_SIZE);
+    config.setDefaultValue(PRODUCER_PENDING_SIZE);
+    config.setDefaultValue(PRODUCER_BLOCK_QUEUE_FULL);
+  }
 }

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

@@ -60,6 +60,8 @@
 import org.apache.hudi.utilities.UtilHelpers;
 import org.apache.hudi.utilities.callback.kafka.HoodieWriteCommitKafkaCallback;
 import org.apache.hudi.utilities.callback.kafka.HoodieWriteCommitKafkaCallbackConfig;
+import org.apache.hudi.utilities.callback.pulsar.HoodieWriteCommitPulsarCallback;
+import org.apache.hudi.utilities.callback.pulsar.HoodieWriteCommitPulsarCallbackConfig;
 import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer.Config;
 import org.apache.hudi.utilities.exception.HoodieDeltaStreamerException;
 import org.apache.hudi.utilities.schema.DelegatingSchemaProvider;
@@ -729,9 +731,16 @@ private HoodieWriteConfig getHoodieClientConfig(Schema schema) {
 
     HoodieWriteConfig config = builder.build();
 
-    // set default value for {@link HoodieWriteCommitKafkaCallbackConfig} if needed.
-    if (config.writeCommitCallbackOn() && HoodieWriteCommitKafkaCallback.class.getName().equals(config.getCallbackClass())) {
-      HoodieWriteCommitKafkaCallbackConfig.setCallbackKafkaConfigIfNeeded(config);
+    if (config.writeCommitCallbackOn()) {
+      // set default value for {@link HoodieWriteCommitKafkaCallbackConfig} if needed.
+      if (HoodieWriteCommitKafkaCallback.class.getName().equals(config.getCallbackClass())) {
+        HoodieWriteCommitKafkaCallbackConfig.setCallbackKafkaConfigIfNeeded(config);
+      }
+
+      // set default value for {@link HoodieWriteCommitPulsarCallbackConfig} if needed.
+      if (HoodieWriteCommitPulsarCallback.class.getName().equals(config.getCallbackClass())) {
+        HoodieWriteCommitPulsarCallbackConfig.setCallbackPulsarConfigIfNeeded(config);
+      }
     }
 
     HoodieClusteringConfig clusteringConfig = HoodieClusteringConfig.from(props);