From b51646dcc76acc68e97dd6a67cc7557e362b590d Mon Sep 17 00:00:00 2001
From: Mathieu <wxhjsxz@126.com>
Date: Fri, 7 Aug 2020 00:03:20 +0800
Subject: [PATCH] [HUDI-1151] Fix NPE when no new data in kafka using
 HoodieDeltaStreamer (#1921)

---
 .../hudi/utilities/deltastreamer/HoodieDeltaStreamer.java       | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java
index fd55e4375..2f6b17e9d 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamer.java
@@ -577,7 +577,7 @@ public class HoodieDeltaStreamer implements Serializable {
             try {
               long start = System.currentTimeMillis();
               Pair<Option<String>, JavaRDD<WriteStatus>> scheduledCompactionInstantAndRDD = deltaSync.syncOnce();
-              if (scheduledCompactionInstantAndRDD.getLeft().isPresent()) {
+              if (null != scheduledCompactionInstantAndRDD && scheduledCompactionInstantAndRDD.getLeft().isPresent()) {
                 LOG.info("Enqueuing new pending compaction instant (" + scheduledCompactionInstantAndRDD.getLeft() + ")");
                 asyncCompactService.enqueuePendingCompaction(new HoodieInstant(State.REQUESTED,
                     HoodieTimeline.COMPACTION_ACTION, scheduledCompactionInstantAndRDD.getLeft().get()));