[HUDI-312] Make docker hdfs cluster ephemeral. This is needed to fix flakiness in integration tests. Also, Fix DeltaStreamer hanging issue due to uncaught exception

docker/compose/docker-compose_hadoop284_hive233_spark231.yml

@@ -21,8 +21,6 @@ services:
     image: apachehudi/hudi-hadoop_2.8.4-namenode:latest
     hostname: namenode
     container_name: namenode
-    volumes:
-      - /tmp/hadoop_name:/hadoop/dfs/name
     environment:
       - CLUSTER_NAME=hudi_hadoop284_hive232_spark231
     ports:
@@ -57,8 +55,6 @@ services:
       retries: 3
     depends_on:
       - namenode
-    volumes:
-      - /tmp/hadoop_data:/hadoop/dfs/data
 
   historyserver:
     image: apachehudi/hudi-hadoop_2.8.4-history:latest

docker/hoodie/hadoop/base/Dockerfile

@@ -24,7 +24,7 @@ USER root
 ENV LANG C.UTF-8
 
 ARG HADOOP_VERSION=2.8.4 
-ARG HADOOP_URL=https://www.apache.org/dist/hadoop/common/hadoop-${HADOOP_VERSION}/hadoop-${HADOOP_VERSION}.tar.gz
+ARG HADOOP_URL=https://archive.apache.org/dist/hadoop/common/hadoop-${HADOOP_VERSION}/hadoop-${HADOOP_VERSION}.tar.gz
 ENV HADOOP_VERSION ${HADOOP_VERSION}
 ENV HADOOP_URL ${HADOOP_URL}
 

docker/setup_demo.sh

@@ -18,16 +18,10 @@
 # limitations under the License.
 ################################################################################
 
-# Create host mount directory and copy
-mkdir -p /tmp/hadoop_name
-mkdir -p /tmp/hadoop_data
-
 WS_ROOT=`dirname $PWD`
 # restart cluster
 HUDI_WS=${WS_ROOT} docker-compose -f compose/docker-compose_hadoop284_hive233_spark231.yml down
 HUDI_WS=${WS_ROOT} docker-compose -f compose/docker-compose_hadoop284_hive233_spark231.yml pull
-rm -rf /tmp/hadoop_data/*
-rm -rf /tmp/hadoop_name/*
 sleep 5
 HUDI_WS=${WS_ROOT} docker-compose -f compose/docker-compose_hadoop284_hive233_spark231.yml up -d
 sleep 15

hudi-integ-test/src/test/java/org/apache/hudi/integ/ITTestHoodieDemo.java

@@ -88,8 +88,8 @@ public class ITTestHoodieDemo extends ITTestBase {
   }
 
   private void setupDemo() throws Exception {
-    List<String> cmds = new ImmutableList.Builder<String>().add("hdfs dfsadmin -safemode wait") // handle NN going into
-                                                                                                // safe mode at times
+    List<String> cmds = new ImmutableList.Builder<String>()
+        .add("hdfs dfsadmin -safemode wait") // handle NN going into safe mode at times
         .add("hdfs dfs -mkdir -p " + HDFS_DATA_DIR)
         .add("hdfs dfs -copyFromLocal -f " + INPUT_BATCH_PATH1 + " " + HDFS_BATCH_PATH1)
         .add("/bin/bash " + DEMO_CONTAINER_SCRIPT).build();

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

@@ -30,6 +30,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Objects;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.hadoop.conf.Configuration;
@@ -251,7 +252,9 @@ public class DeltaSync implements Serializable {
         } else {
           throw new HoodieDeltaStreamerException(
               "Unable to find previous checkpoint. Please double check if this table "
-                  + "was indeed built via delta streamer ");
+                  + "was indeed built via delta streamer. Last Commit :" + lastCommit + ", Instants :"
+                  + commitTimelineOpt.get().getInstants().collect(Collectors.toList()) + ", CommitMetadata="
+                  + commitMetadata.toJsonString());
         }
       }
     } else {

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

@@ -117,9 +117,15 @@ public class HoodieDeltaStreamer implements Serializable {
       log.info("Delta Sync shutting down");
     } else {
       log.info("Delta Streamer running only single round");
-      deltaSyncService.getDeltaSync().syncOnce();
-      deltaSyncService.close();
-      log.info("Shut down deltastreamer");
+      try {
+        deltaSyncService.getDeltaSync().syncOnce();
+      } catch (Exception ex) {
+        log.error("Got error running delta sync once. Shutting down", ex);
+        throw ex;
+      } finally {
+        deltaSyncService.close();
+        log.info("Shut down deltastreamer");
+      }
     }
   }