Add unique id to CopyEntity instead of computing it from serialised json string

Author: thisisArjit

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/CopyEntity.java

@@ -24,7 +24,6 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
-import java.util.UUID;
 
 import lombok.AccessLevel;
 import lombok.AllArgsConstructor;
@@ -55,8 +54,6 @@ public class CopyEntity implements HasGuid {
 
   public static final Gson GSON = GsonInterfaceAdapter.getGson(Object.class);
 
-  private final Guid id = new Guid(UUID.randomUUID().toString().getBytes());
-
   /**
    * File set this file belongs to. {@link CopyEntity}s in the same fileSet and originating from the same
    * {@link CopyableDataset} will be treated as a unit: they will be published nearly atomically, and a notification
@@ -69,7 +66,7 @@ public class CopyEntity implements HasGuid {
 
   @Override
   public Guid guid() throws IOException {
-    return id;
+    return Guid.fromStrings(toString());
   }
 
   /**

gobblin-data-management/src/main/java/org/apache/gobblin/data/management/copy/CopySource.java

@@ -19,10 +19,12 @@
 
 import java.io.IOException;
 import java.net.URI;
+import java.nio.charset.StandardCharsets;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
@@ -54,6 +56,7 @@
 import org.apache.gobblin.data.management.copy.extractor.FileAwareInputStreamExtractor;
 import org.apache.gobblin.data.management.copy.prioritization.FileSetComparator;
 import org.apache.gobblin.data.management.copy.publisher.CopyEventSubmitterHelper;
+import org.apache.gobblin.data.management.copy.recovery.RecoveryHelper;
 import org.apache.gobblin.data.management.copy.replication.ConfigBasedDataset;
 import org.apache.gobblin.data.management.copy.splitter.DistcpFileSplitter;
 import org.apache.gobblin.data.management.copy.watermark.CopyableFileWatermarkGenerator;
@@ -372,6 +375,10 @@ public Void call() {
 
         Extract extract = new Extract(Extract.TableType.SNAPSHOT_ONLY, CopyConfiguration.COPY_PREFIX, extractId);
         List<WorkUnit> workUnitsForPartition = Lists.newArrayList();
+        // use random guid for workunits if recovery is disabled. Getting deterministic guid is expensive as it
+        // serialises entire workunit to generate guid
+        boolean useRandomGuidForWorkUnit = !isRecoveryEnabled(state);
+        log.info("Using " + (useRandomGuidForWorkUnit ? "random" : "deterministic") + " guids for workunits");
 
         long fileSize;
         for (CopyEntity copyEntity : fileSet.getFiles()) {
@@ -400,7 +407,7 @@ public Void call() {
           workUnit.setProp(SlaEventKeys.PARTITION_KEY, copyEntity.getFileSet());
           setWorkUnitWeight(workUnit, copyEntity, minWorkUnitWeight);
           setWorkUnitWatermark(workUnit, watermarkGenerator, copyEntity);
-          computeAndSetWorkUnitGuid(workUnit);
+          computeAndSetWorkUnitGuid(workUnit, copyEntity, useRandomGuidForWorkUnit);
           addLineageInfo(copyEntity, workUnit);
           if (copyEntity instanceof CopyableFile) {
             CopyableFile castedCopyEntity = (CopyableFile) copyEntity;
@@ -427,6 +434,10 @@ public Void call() {
       }
     }
 
+    private static boolean isRecoveryEnabled(State state) {
+      return state.contains(RecoveryHelper.PERSIST_DIR_KEY);
+    }
+
     private void setWorkUnitWatermark(WorkUnit workUnit, Optional<CopyableFileWatermarkGenerator> watermarkGenerator,
         CopyEntity copyEntity)
         throws IOException {
@@ -514,11 +525,12 @@ private static void setWorkUnitWeight(WorkUnit workUnit, CopyEntity copyEntity,
     workUnit.setProp(WORK_UNIT_WEIGHT, Long.toString(weight));
   }
 
-  private static void computeAndSetWorkUnitGuid(WorkUnit workUnit)
+  private static void computeAndSetWorkUnitGuid(WorkUnit workUnit, CopyEntity copyEntity, boolean useRandomGuidForWorkUnit)
       throws IOException {
     Guid guid = Guid.fromStrings(workUnit.contains(ConfigurationKeys.CONVERTER_CLASSES_KEY) ? workUnit
         .getProp(ConfigurationKeys.CONVERTER_CLASSES_KEY) : "");
-    setWorkUnitGuid(workUnit, guid.append(deserializeCopyEntity(workUnit)));
+    Guid copyEntityGuid = useRandomGuidForWorkUnit ? new Guid(UUID.randomUUID().toString().getBytes(StandardCharsets.UTF_8)) : copyEntity.guid();
+    setWorkUnitGuid(workUnit, guid.append(copyEntityGuid));
   }
 
   /**