[Stacked on 6386] Fixing DebeziumSource to properly commit offsets; (apache#6416)

Author: Alexey Kudinkin

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

@@ -340,6 +340,7 @@ public Pair<Option<String>, JavaRDD<WriteStatus>> syncOnce() throws IOException
 
     metrics.updateDeltaStreamerSyncMetrics(System.currentTimeMillis());
 
+    // TODO revisit (too early to unpersist)
     // Clear persistent RDDs
     jssc.getPersistentRDDs().values().forEach(JavaRDD::unpersist);
     return result;

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

@@ -832,7 +832,7 @@ protected Boolean onInitializingWriteClient(SparkRDDWriteClient writeClient) {
      * Close all resources.
      */
     public void close() {
-      if (null != deltaSync) {
+      if (deltaSync != null) {
         deltaSync.close();
       }
     }

hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/debezium/DebeziumSource.java

@@ -239,5 +239,13 @@ private static Dataset<Row> convertArrayColumnsToString(Dataset<Row> dataset) {
 
     return dataset;
   }
+
+  @Override
+  public void onCommit(String lastCkptStr) {
+    if (this.props.getBoolean(KafkaOffsetGen.Config.ENABLE_KAFKA_COMMIT_OFFSET.key(),
+        KafkaOffsetGen.Config.ENABLE_KAFKA_COMMIT_OFFSET.defaultValue())) {
+      offsetGen.commitOffsetToKafka(lastCkptStr);
+    }
+  }
 }
 

hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaOffsetGen.java

@@ -110,11 +110,13 @@ public static OffsetRange[] computeOffsetRanges(Map<TopicPartition, Long> fromOf
       Comparator<OffsetRange> byPartition = Comparator.comparing(OffsetRange::partition);
 
       // Create initial offset ranges for each 'to' partition, with from = to offsets.
-      OffsetRange[] ranges = new OffsetRange[toOffsetMap.size()];
-      toOffsetMap.keySet().stream().map(tp -> {
+      OffsetRange[] ranges = toOffsetMap.keySet().stream().map(tp -> {
         long fromOffset = fromOffsetMap.getOrDefault(tp, 0L);
         return OffsetRange.create(tp, fromOffset, fromOffset);
-      }).sorted(byPartition).collect(Collectors.toList()).toArray(ranges);
+      })
+          .sorted(byPartition)
+          .collect(Collectors.toList())
+          .toArray(new OffsetRange[toOffsetMap.size()]);
 
       long allocedEvents = 0;
       Set<Integer> exhaustedPartitions = new HashSet<>();
@@ -290,6 +292,7 @@ public OffsetRange[] getNextOffsetRanges(Option<String> lastCheckpointStr, long
       numEvents = sourceLimit;
     }
 
+    // TODO(HUDI-4625) remove
     if (numEvents < toOffsets.size()) {
       throw new HoodieException("sourceLimit should not be less than the number of kafka partitions");
     }
@@ -309,6 +312,7 @@ private List<PartitionInfo> fetchPartitionInfos(KafkaConsumer consumer, String t
 
     List<PartitionInfo> partitionInfos;
     do {
+      // TODO(HUDI-4625) cleanup, introduce retrying client
       partitionInfos = consumer.partitionsFor(topicName);
       try {
         TimeUnit.SECONDS.sleep(10);