HUDI-479: Eliminate or Minimize use of Guava if possible (#1159)

hudi-common/src/main/java/org/apache/hudi/avro/MercifulJsonConverter.java

@@ -23,7 +23,6 @@ import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.exception.HoodieIOException;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.collect.ImmutableMap;
 import org.apache.avro.Schema;
 import org.apache.avro.Schema.Type;
 import org.apache.avro.generic.GenericData;
@@ -32,6 +31,7 @@ import org.apache.avro.generic.GenericRecord;
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -49,13 +49,22 @@ public class MercifulJsonConverter {
    * Build type processor map for each avro type.
    */
   private static Map<Schema.Type, JsonToAvroFieldProcessor> getFieldTypeProcessors() {
-    return new ImmutableMap.Builder<Type, JsonToAvroFieldProcessor>().put(Type.STRING, generateStringTypeHandler())
-        .put(Type.BOOLEAN, generateBooleanTypeHandler()).put(Type.DOUBLE, generateDoubleTypeHandler())
-        .put(Type.FLOAT, generateFloatTypeHandler()).put(Type.INT, generateIntTypeHandler())
-        .put(Type.LONG, generateLongTypeHandler()).put(Type.ARRAY, generateArrayTypeHandler())
-        .put(Type.RECORD, generateRecordTypeHandler()).put(Type.ENUM, generateEnumTypeHandler())
-        .put(Type.MAP, generateMapTypeHandler()).put(Type.BYTES, generateBytesTypeHandler())
-        .put(Type.FIXED, generateFixedTypeHandler()).build();
+    return Collections.unmodifiableMap(new HashMap<Schema.Type, JsonToAvroFieldProcessor>() {
+      {
+        put(Type.STRING, generateStringTypeHandler());
+        put(Type.BOOLEAN, generateBooleanTypeHandler());
+        put(Type.DOUBLE, generateDoubleTypeHandler());
+        put(Type.FLOAT, generateFloatTypeHandler());
+        put(Type.INT, generateIntTypeHandler());
+        put(Type.LONG, generateLongTypeHandler());
+        put(Type.ARRAY, generateArrayTypeHandler());
+        put(Type.RECORD, generateRecordTypeHandler());
+        put(Type.ENUM, generateEnumTypeHandler());
+        put(Type.MAP, generateMapTypeHandler());
+        put(Type.BYTES, generateBytesTypeHandler());
+        put(Type.FIXED, generateFixedTypeHandler());
+      }
+    });
   }
 
   /**

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieRecord.java

@@ -18,10 +18,9 @@
 
 package org.apache.hudi.common.model;
 
+import org.apache.hudi.common.util.CollectionUtils;
 import org.apache.hudi.common.util.Option;
 
-import com.google.common.collect.ImmutableList;
-
 import java.io.Serializable;
 import java.util.List;
 import java.util.Objects;
@@ -38,8 +37,8 @@ public class HoodieRecord<T extends HoodieRecordPayload> implements Serializable
   public static String FILENAME_METADATA_FIELD = "_hoodie_file_name";
 
   public static final List<String> HOODIE_META_COLUMNS =
-      new ImmutableList.Builder<String>().add(COMMIT_TIME_METADATA_FIELD).add(COMMIT_SEQNO_METADATA_FIELD)
-          .add(RECORD_KEY_METADATA_FIELD).add(PARTITION_PATH_METADATA_FIELD).add(FILENAME_METADATA_FIELD).build();
+      CollectionUtils.createImmutableList(COMMIT_TIME_METADATA_FIELD, COMMIT_SEQNO_METADATA_FIELD,
+          RECORD_KEY_METADATA_FIELD, PARTITION_PATH_METADATA_FIELD, FILENAME_METADATA_FIELD);
 
   /**
    * Identifies the record across the table.

hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java

@@ -26,13 +26,13 @@ import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
 import org.apache.hudi.common.table.timeline.HoodieArchivedTimeline;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.util.ConsistencyGuardConfig;
-import org.apache.hudi.common.util.FailSafeConsistencyGuard;
 import org.apache.hudi.common.util.FSUtils;
+import org.apache.hudi.common.util.FailSafeConsistencyGuard;
 import org.apache.hudi.common.util.NoOpConsistencyGuard;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.ValidationUtils;
-import org.apache.hudi.exception.TableNotFoundException;
 import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.TableNotFoundException;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
@@ -246,7 +246,7 @@ public class HoodieTableMetaClient implements Serializable {
   /**
    * Return raw file-system.
    * 
-   * @return
+   * @return fs
    */
   public FileSystem getRawFs() {
     return getFs().getFileSystem();

hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java

@@ -26,7 +26,6 @@ import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.ValidationUtils;
 import org.apache.hudi.exception.HoodieIOException;
 
-import com.google.common.collect.ImmutableSet;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.Path;
@@ -38,8 +37,10 @@ import java.io.IOException;
 import java.io.Serializable;
 import java.text.SimpleDateFormat;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Date;
 import java.util.HashSet;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
@@ -106,13 +107,11 @@ public class HoodieActiveTimeline extends HoodieDefaultTimeline {
   }
 
   public HoodieActiveTimeline(HoodieTableMetaClient metaClient) {
-    this(metaClient, new ImmutableSet.Builder<String>().addAll(VALID_EXTENSIONS_IN_ACTIVE_TIMELINE).build());
+    this(metaClient, Collections.unmodifiableSet(VALID_EXTENSIONS_IN_ACTIVE_TIMELINE));
   }
 
   public HoodieActiveTimeline(HoodieTableMetaClient metaClient, boolean applyLayoutFilter) {
-    this(metaClient,
-        new ImmutableSet.Builder<String>()
-            .addAll(VALID_EXTENSIONS_IN_ACTIVE_TIMELINE).build(), applyLayoutFilter);
+    this(metaClient, Collections.unmodifiableSet(VALID_EXTENSIONS_IN_ACTIVE_TIMELINE), applyLayoutFilter);
   }
 
   /**
@@ -166,7 +165,7 @@ public class HoodieActiveTimeline extends HoodieDefaultTimeline {
 
   public void deleteCompactionRequested(HoodieInstant instant) {
     ValidationUtils.checkArgument(instant.isRequested());
-    ValidationUtils.checkArgument(instant.getAction().equals(HoodieTimeline.COMPACTION_ACTION));
+    ValidationUtils.checkArgument(Objects.equals(instant.getAction(), HoodieTimeline.COMPACTION_ACTION));
     deleteInstantFile(instant);
   }
 

hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieArchivedTimeline.java

@@ -31,6 +31,7 @@ import org.apache.hudi.common.table.log.HoodieLogFormat;
 import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.exception.HoodieIOException;
+
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 

hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieDefaultTimeline.java

@@ -20,11 +20,11 @@ package org.apache.hudi.common.table.timeline;
 
 import org.apache.hudi.common.table.HoodieTimeline;
 import org.apache.hudi.common.table.timeline.HoodieInstant.State;
+import org.apache.hudi.common.util.CollectionUtils;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.StringUtils;
 import org.apache.hudi.exception.HoodieException;
 
-import com.google.common.collect.Sets;
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
@@ -113,7 +113,7 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
 
   @Override
   public HoodieTimeline getCommitsAndCompactionTimeline() {
-    Set<String> validActions = Sets.newHashSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, COMPACTION_ACTION);
+    Set<String> validActions = CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, COMPACTION_ACTION);
     return new HoodieDefaultTimeline(instants.stream().filter(s -> validActions.contains(s.getAction())), details);
   }
 
@@ -145,7 +145,7 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
    * Get all instants (commits, delta commits) that produce new data, in the active timeline.
    */
   public HoodieTimeline getCommitsTimeline() {
-    return getTimelineOfActions(Sets.newHashSet(COMMIT_ACTION, DELTA_COMMIT_ACTION));
+    return getTimelineOfActions(CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION));
   }
 
   /**
@@ -153,8 +153,8 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
    * timeline.
    */
   public HoodieTimeline getAllCommitsTimeline() {
-    return getTimelineOfActions(Sets.newHashSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, CLEAN_ACTION, COMPACTION_ACTION,
-            SAVEPOINT_ACTION, ROLLBACK_ACTION));
+    return getTimelineOfActions(CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION,
+            CLEAN_ACTION, COMPACTION_ACTION, SAVEPOINT_ACTION, ROLLBACK_ACTION));
   }
 
   /**

hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieInstant.java

@@ -19,9 +19,9 @@
 package org.apache.hudi.common.table.timeline;
 
 import org.apache.hudi.common.table.HoodieTimeline;
+import org.apache.hudi.common.util.CollectionUtils;
 import org.apache.hudi.common.util.FSUtils;
 
-import com.google.common.collect.ImmutableMap;
 import org.apache.hadoop.fs.FileStatus;
 
 import java.io.Serializable;
@@ -41,8 +41,8 @@ public class HoodieInstant implements Serializable, Comparable<HoodieInstant> {
    * A COMPACTION action eventually becomes COMMIT when completed. So, when grouping instants
    * for state transitions, this needs to be taken into account
    */
-  private static final Map<String, String> COMPARABLE_ACTIONS = new ImmutableMap.Builder<String, String>()
-      .put(HoodieTimeline.COMPACTION_ACTION, HoodieTimeline.COMMIT_ACTION).build();
+  private static final Map<String, String> COMPARABLE_ACTIONS =
+      CollectionUtils.createImmutableMap(HoodieTimeline.COMPACTION_ACTION, HoodieTimeline.COMMIT_ACTION);
 
   public static final Comparator<HoodieInstant> ACTION_COMPARATOR =
       Comparator.comparing(instant -> getComparableAction(instant.getAction()));

hudi-common/src/main/java/org/apache/hudi/common/table/view/IncrementalTimelineSyncFileSystemView.java

@@ -24,8 +24,8 @@ import org.apache.hudi.avro.model.HoodieRestoreMetadata;
 import org.apache.hudi.avro.model.HoodieRollbackMetadata;
 import org.apache.hudi.common.model.CompactionOperation;
 import org.apache.hudi.common.model.FileSlice;
-import org.apache.hudi.common.model.HoodieCommitMetadata;
 import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
 import org.apache.hudi.common.model.HoodieFileGroup;
 import org.apache.hudi.common.model.HoodieLogFile;
 import org.apache.hudi.common.table.HoodieTimeline;

hudi-common/src/main/java/org/apache/hudi/common/table/view/RemoteHoodieTableFileSystemView.java

@@ -26,8 +26,8 @@ import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.HoodieTimeline;
 import org.apache.hudi.common.table.SyncableFileSystemView;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.table.timeline.dto.CompactionOpDTO;
 import org.apache.hudi.common.table.timeline.dto.BaseFileDTO;
+import org.apache.hudi.common.table.timeline.dto.CompactionOpDTO;
 import org.apache.hudi.common.table.timeline.dto.FileGroupDTO;
 import org.apache.hudi.common.table.timeline.dto.FileSliceDTO;
 import org.apache.hudi.common.table.timeline.dto.InstantDTO;

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

@@ -28,7 +28,6 @@ import org.apache.hudi.avro.model.HoodieSavepointMetadata;
 import org.apache.hudi.avro.model.HoodieSavepointPartitionMetadata;
 import org.apache.hudi.common.HoodieRollbackStat;
 
-import com.google.common.collect.ImmutableMap;
 import org.apache.avro.file.DataFileReader;
 import org.apache.avro.file.DataFileWriter;
 import org.apache.avro.file.FileReader;
@@ -42,6 +41,7 @@ import org.apache.avro.specific.SpecificRecordBase;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -54,18 +54,18 @@ public class AvroUtils {
 
   public static HoodieRestoreMetadata convertRestoreMetadata(String startRestoreTime, Option<Long> durationInMs,
       List<String> commits, Map<String, List<HoodieRollbackStat>> commitToStats) {
-    ImmutableMap.Builder<String, List<HoodieRollbackMetadata>> commitToStatBuilder = ImmutableMap.builder();
+    Map<String, List<HoodieRollbackMetadata>> commitToStatsMap = new HashMap<>();
     for (Map.Entry<String, List<HoodieRollbackStat>> commitToStat : commitToStats.entrySet()) {
-      commitToStatBuilder.put(commitToStat.getKey(),
+      commitToStatsMap.put(commitToStat.getKey(),
           Collections.singletonList(convertRollbackMetadata(startRestoreTime, durationInMs, commits, commitToStat.getValue())));
     }
     return new HoodieRestoreMetadata(startRestoreTime, durationInMs.orElseGet(() -> -1L), commits,
-        commitToStatBuilder.build(), DEFAULT_VERSION);
+      Collections.unmodifiableMap(commitToStatsMap), DEFAULT_VERSION);
   }
 
   public static HoodieRollbackMetadata convertRollbackMetadata(String startRollbackTime, Option<Long> durationInMs,
       List<String> commits, List<HoodieRollbackStat> rollbackStats) {
-    ImmutableMap.Builder<String, HoodieRollbackPartitionMetadata> partitionMetadataBuilder = ImmutableMap.builder();
+    Map<String, HoodieRollbackPartitionMetadata> partitionMetadataBuilder = new HashMap<>();
     int totalDeleted = 0;
     for (HoodieRollbackStat stat : rollbackStats) {
       HoodieRollbackPartitionMetadata metadata = new HoodieRollbackPartitionMetadata(stat.getPartitionPath(),
@@ -75,18 +75,18 @@ public class AvroUtils {
     }
 
     return new HoodieRollbackMetadata(startRollbackTime, durationInMs.orElseGet(() -> -1L), totalDeleted, commits,
-        partitionMetadataBuilder.build(), DEFAULT_VERSION);
+      Collections.unmodifiableMap(partitionMetadataBuilder), DEFAULT_VERSION);
   }
 
   public static HoodieSavepointMetadata convertSavepointMetadata(String user, String comment,
       Map<String, List<String>> latestFiles) {
-    ImmutableMap.Builder<String, HoodieSavepointPartitionMetadata> partitionMetadataBuilder = ImmutableMap.builder();
+    Map<String, HoodieSavepointPartitionMetadata> partitionMetadataBuilder = new HashMap<>();
     for (Map.Entry<String, List<String>> stat : latestFiles.entrySet()) {
       HoodieSavepointPartitionMetadata metadata = new HoodieSavepointPartitionMetadata(stat.getKey(), stat.getValue());
       partitionMetadataBuilder.put(stat.getKey(), metadata);
     }
-    return new HoodieSavepointMetadata(user, System.currentTimeMillis(), comment, partitionMetadataBuilder.build(),
-        DEFAULT_VERSION);
+    return new HoodieSavepointMetadata(user, System.currentTimeMillis(), comment,
+      Collections.unmodifiableMap(partitionMetadataBuilder), DEFAULT_VERSION);
   }
 
   public static Option<byte[]> serializeCompactionPlan(HoodieCompactionPlan compactionWorkload) throws IOException {

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

@@ -28,10 +28,10 @@ import org.apache.hudi.common.versioning.clean.CleanMetadataMigrator;
 import org.apache.hudi.common.versioning.clean.CleanV1MigrationHandler;
 import org.apache.hudi.common.versioning.clean.CleanV2MigrationHandler;
 
-import com.google.common.collect.ImmutableMap;
-
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 public class CleanerUtils {
   public static final Integer CLEAN_METADATA_VERSION_1 = CleanV1MigrationHandler.VERSION;
@@ -40,14 +40,14 @@ public class CleanerUtils {
 
   public static HoodieCleanMetadata convertCleanMetadata(HoodieTableMetaClient metaClient,
       String startCleanTime, Option<Long> durationInMs, List<HoodieCleanStat> cleanStats) {
-    ImmutableMap.Builder<String, HoodieCleanPartitionMetadata> partitionMetadataBuilder = ImmutableMap.builder();
+    Map<String, HoodieCleanPartitionMetadata> partitionMetadataMap = new HashMap<>();
     int totalDeleted = 0;
     String earliestCommitToRetain = null;
     for (HoodieCleanStat stat : cleanStats) {
       HoodieCleanPartitionMetadata metadata =
           new HoodieCleanPartitionMetadata(stat.getPartitionPath(), stat.getPolicy().name(),
               stat.getDeletePathPatterns(), stat.getSuccessDeleteFiles(), stat.getFailedDeleteFiles());
-      partitionMetadataBuilder.put(stat.getPartitionPath(), metadata);
+      partitionMetadataMap.put(stat.getPartitionPath(), metadata);
       totalDeleted += stat.getSuccessDeleteFiles().size();
       if (earliestCommitToRetain == null) {
         // This will be the same for all partitions
@@ -56,8 +56,7 @@ public class CleanerUtils {
     }
 
     return new HoodieCleanMetadata(startCleanTime,
-        durationInMs.orElseGet(() -> -1L), totalDeleted, earliestCommitToRetain,
-        partitionMetadataBuilder.build(), CLEAN_METADATA_VERSION_2);
+        durationInMs.orElseGet(() -> -1L), totalDeleted, earliestCommitToRetain, partitionMetadataMap, CLEAN_METADATA_VERSION_2);
   }
 
   /**

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

@@ -0,0 +1,111 @@
+/*
+ * 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 org.apache.hudi.common.util.collection.Pair;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class CollectionUtils {
+  /**
+   * Determines whether two iterators contain equal elements in the same order. More specifically,
+   * this method returns {@code true} if {@code iterator1} and {@code iterator2} contain the same
+   * number of elements and every element of {@code iterator1} is equal to the corresponding element
+   * of {@code iterator2}.
+   *
+   * <p>Note that this will modify the supplied iterators, since they will have been advanced some
+   * number of elements forward.
+   */
+  public static boolean elementsEqual(Iterator<?> iterator1, Iterator<?> iterator2) {
+    while (iterator1.hasNext()) {
+      if (!iterator2.hasNext()) {
+        return false;
+      }
+      Object o1 = iterator1.next();
+      Object o2 = iterator2.next();
+      if (!Objects.equals(o1, o2)) {
+        return false;
+      }
+    }
+    return !iterator2.hasNext();
+  }
+
+  @SafeVarargs
+  public static <T> Set<T> createSet(final T... elements) {
+    return Stream.of(elements).collect(Collectors.toSet());
+  }
+
+  public static <K,V> Map<K, V> createImmutableMap(final K key, final V value) {
+    return Collections.unmodifiableMap(Collections.singletonMap(key, value));
+  }
+
+  @SafeVarargs
+  public static <T> List<T> createImmutableList(final T... elements) {
+    return Collections.unmodifiableList(Stream.of(elements).collect(Collectors.toList()));
+  }
+
+  public static <K,V> Map<K,V> createImmutableMap(final Map<K,V> map) {
+    return Collections.unmodifiableMap(map);
+  }
+
+  @SafeVarargs
+  public static <K,V> Map<K,V> createImmutableMap(final Pair<K,V>... elements) {
+    Map<K,V> map = new HashMap<>();
+    for (Pair<K,V> pair: elements) {
+      map.put(pair.getLeft(), pair.getRight());
+    }
+    return Collections.unmodifiableMap(map);
+  }
+
+  @SafeVarargs
+  public static <T> Set<T> createImmutableSet(final T... elements) {
+    return Collections.unmodifiableSet(createSet(elements));
+  }
+
+  public static <T> Set<T> createImmutableSet(final Set<T> set) {
+    return Collections.unmodifiableSet(set);
+  }
+
+  public static <T> List<T> createImmutableList(final List<T> list) {
+    return Collections.unmodifiableList(list);
+  }
+
+  private static Object[] checkElementsNotNull(Object... array) {
+    return checkElementsNotNull(array, array.length);
+  }
+
+  private static Object[] checkElementsNotNull(Object[] array, int length) {
+    for (int i = 0; i < length; i++) {
+      checkElementNotNull(array[i], i);
+    }
+    return array;
+  }
+
+  private static Object checkElementNotNull(Object element, int index) {
+    return Objects.requireNonNull(element, "Element is null at index " + index);
+  }
+}

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

@@ -46,8 +46,8 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.Objects;
 import java.util.Map.Entry;
+import java.util.Objects;
 import java.util.UUID;
 import java.util.function.Function;
 import java.util.regex.Matcher;
@@ -213,7 +213,7 @@ public class FSUtils {
    * @param basePathStr Base-Path
    * @param consumer Callback for processing
    * @param excludeMetaFolder Exclude .hoodie folder
-   * @throws IOException
+   * @throws IOException -
    */
   static void processFiles(FileSystem fs, String basePathStr, Function<FileStatus, Boolean> consumer,
       boolean excludeMetaFolder) throws IOException {

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

@@ -18,7 +18,10 @@
 
 package org.apache.hudi.common.util;
 
+import javax.annotation.Nullable;
+
 import java.io.ByteArrayOutputStream;
+import java.io.Closeable;
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
@@ -91,4 +94,26 @@ public class FileIOUtils {
     out.flush();
     out.close();
   }
+
+  /**
+   * Closes a {@link Closeable}, with control over whether an {@code IOException} may be thrown.
+   * @param closeable the {@code Closeable} object to be closed, or null,
+   *      in which case this method does nothing.
+   * @param swallowIOException if true, don't propagate IO exceptions thrown by the {@code close} methods.
+   *
+   * @throws IOException if {@code swallowIOException} is false and {@code close} throws an {@code IOException}.
+   */
+  public static void close(@Nullable Closeable closeable, boolean swallowIOException)
+      throws IOException {
+    if (closeable == null) {
+      return;
+    }
+    try {
+      closeable.close();
+    } catch (IOException e) {
+      if (!swallowIOException) {
+        throw e;
+      }
+    }
+  }
 }

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

@@ -21,14 +21,21 @@ package org.apache.hudi.common.util;
 import org.apache.hudi.common.model.HoodieRecordPayload;
 import org.apache.hudi.exception.HoodieException;
 
-import com.google.common.reflect.ClassPath;
-import com.google.common.reflect.ClassPath.ClassInfo;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
 
+import java.io.File;
 import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Enumeration;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.stream.Stream;
 
 /**
@@ -36,6 +43,8 @@ import java.util.stream.Stream;
  */
 public class ReflectionUtils {
 
+  private static final Logger LOG = LogManager.getLogger(ReflectionUtils.class);
+
   private static Map<String, Class<?>> clazzCache = new HashMap<>();
 
   private static Class<?> getClass(String clazzName) {
@@ -90,16 +99,58 @@ public class ReflectionUtils {
   }
 
   /**
-   * Return stream of top level class names in the same class path as passed-in class.
-   * 
-   * @param clazz
+   * Scans all classes accessible from the context class loader
+   * which belong to the given package and subpackages.
+   *
+   * @param clazz class
+   * @return Stream of Class names in package
    */
-  public static Stream<String> getTopLevelClassesInClasspath(Class clazz) {
+  public static Stream<String> getTopLevelClassesInClasspath(Class<?> clazz) {
+    ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+    String packageName = clazz.getPackage().getName();
+    String path = packageName.replace('.', '/');
+    Enumeration<URL> resources = null;
     try {
-      ClassPath classPath = ClassPath.from(clazz.getClassLoader());
-      return classPath.getTopLevelClasses().stream().map(ClassInfo::getName);
+      resources = classLoader.getResources(path);
     } catch (IOException e) {
-      throw new RuntimeException("Got exception while dumping top level classes", e);
+      LOG.error("Unable to fetch Resources in package " + e.getMessage());
     }
+    List<File> directories = new ArrayList<>();
+    while (Objects.requireNonNull(resources).hasMoreElements()) {
+      URL resource = resources.nextElement();
+      try {
+        directories.add(new File(resource.toURI()));
+      } catch (URISyntaxException e) {
+        LOG.error("Unable to get " + e.getMessage());
+      }
+    }
+    List<String> classes = new ArrayList<>();
+    for (File directory : directories) {
+      classes.addAll(findClasses(directory, packageName));
+    }
+    return classes.stream();
+  }
+
+  /**
+   * Recursive method used to find all classes in a given directory and subdirs.
+   *
+   * @param directory   The base directory
+   * @param packageName The package name for classes found inside the base directory
+   * @return classes in the package
+   */
+  private static List<String> findClasses(File directory, String packageName) {
+    List<String> classes = new ArrayList<>();
+    if (!directory.exists()) {
+      return classes;
+    }
+    File[] files = directory.listFiles();
+    for (File file : Objects.requireNonNull(files)) {
+      if (file.isDirectory()) {
+        classes.addAll(findClasses(file, packageName + "." + file.getName()));
+      } else if (file.getName().endsWith(".class")) {
+        classes.add(packageName + '.' + file.getName().substring(0, file.getName().length() - 6));
+      }
+    }
+    return classes;
   }
 }