Renames fork => dynamic split in the Dataflow Java SDK.

----Release Notes----
[]
-------------
Created by MOE: http://code.google.com/p/moe-java
MOE_MIGRATED_REVID=89358962

Author: jkff

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/dataflow/BasicSerializableSourceFormat.java

@@ -308,7 +308,7 @@ public Reader.Progress getProgress() {
     }
 
     @Override
-    public Reader.ForkResult requestFork(Reader.ForkRequest request) {
+    public Reader.DynamicSplitResult requestDynamicSplit(Reader.DynamicSplitRequest request) {
       return null;
     }
   }

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/worker/BigQueryReader.java

@@ -96,8 +96,8 @@ public Progress getProgress() {
     }
 
     @Override
-    public ForkResult requestFork(ForkRequest forkRequest) {
-      // For now fork is not supported because this source
+    public DynamicSplitResult requestDynamicSplit(DynamicSplitRequest splitRequest) {
+      // For now dynamic splitting is not supported because this source
       // is used only when an entire table needs to be read by each worker (used
       // as a side input for instance).
       throw new UnsupportedOperationException();

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/worker/DataflowWorkProgressUpdater.java

@@ -18,7 +18,7 @@
 
 import static com.google.cloud.dataflow.sdk.runners.worker.DataflowWorker.buildStatus;
 import static com.google.cloud.dataflow.sdk.runners.worker.DataflowWorker.uniqueId;
-import static com.google.cloud.dataflow.sdk.runners.worker.SourceTranslationUtils.toForkRequest;
+import static com.google.cloud.dataflow.sdk.runners.worker.SourceTranslationUtils.toDynamicSplitRequest;
 import static com.google.cloud.dataflow.sdk.util.TimeUtil.fromCloudDuration;
 import static com.google.cloud.dataflow.sdk.util.TimeUtil.fromCloudTime;
 import static com.google.cloud.dataflow.sdk.util.TimeUtil.toCloudDuration;
@@ -79,15 +79,15 @@ protected long getWorkUnitLeaseExpirationTimestamp() {
   @Override
   protected void reportProgressHelper() throws Exception {
     WorkItemStatus status = buildStatus(workItem, false/*completed*/, worker.getOutputCounters(),
-        worker.getOutputMetrics(), options, worker.getWorkerProgress(), forkResultToReport,
+        worker.getOutputMetrics(), options, worker.getWorkerProgress(), dynamicSplitResultToReport,
         null/*sourceOperationResponse*/, null/*errors*/,
         getNextReportIndex());
     status.setRequestedLeaseDuration(toCloudDuration(Duration.millis(requestedLeaseDurationMs)));
 
     WorkItemServiceState result = workUnitClient.reportWorkItemStatus(status);
     if (result != null) {
       // Resets state after a successful progress report.
-      forkResultToReport = null;
+      dynamicSplitResultToReport = null;
       nextReportIndex++;
 
       progressReportIntervalMs = nextProgressReportInterval(
@@ -96,8 +96,9 @@ protected void reportProgressHelper() throws Exception {
 
       ApproximateProgress suggestedStopPoint = result.getSuggestedStopPoint();
       if (suggestedStopPoint != null) {
-        LOG.info("Proposing fork of work unit {} at {}", workString(), suggestedStopPoint);
-        forkResultToReport = worker.requestFork(toForkRequest(suggestedStopPoint));
+        LOG.info("Proposing dynamic split of work unit {} at {}", workString(), suggestedStopPoint);
+        dynamicSplitResultToReport = worker.requestDynamicSplit(
+            toDynamicSplitRequest(suggestedStopPoint));
       }
     }
   }

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/worker/DataflowWorker.java

@@ -211,7 +211,7 @@ private void reportStatus(DataflowWorkerHarnessOptions options, String status, W
   static WorkItemStatus buildStatus(WorkItem workItem, boolean completed,
       @Nullable CounterSet counters, @Nullable Collection<Metric<?>> metrics,
       DataflowWorkerHarnessOptions options, @Nullable Reader.Progress progress,
-      @Nullable Reader.ForkResult forkResult,
+      @Nullable Reader.DynamicSplitResult dynamicSplitResult,
       @Nullable SourceFormat.OperationResponse operationResponse, @Nullable List<Status> errors,
       long finalReportIndex) {
     WorkItemStatus status = new WorkItemStatus();
@@ -256,11 +256,13 @@ static WorkItemStatus buildStatus(WorkItem workItem, boolean completed,
     if (progress != null) {
       status.setProgress(readerProgressToCloudProgress(progress));
     }
-    if (forkResult instanceof Reader.ForkResultWithPosition) {
-      Reader.ForkResultWithPosition asPosition = (Reader.ForkResultWithPosition) forkResult;
+    if (dynamicSplitResult instanceof Reader.DynamicSplitResultWithPosition) {
+      Reader.DynamicSplitResultWithPosition asPosition =
+          (Reader.DynamicSplitResultWithPosition) dynamicSplitResult;
       status.setStopPosition(toCloudPosition(asPosition.getAcceptedPosition()));
-    } else if (forkResult != null) {
-      throw new IllegalArgumentException("Unexpected type of fork result: " + forkResult);
+    } else if (dynamicSplitResult != null) {
+      throw new IllegalArgumentException(
+          "Unexpected type of dynamic split result: " + dynamicSplitResult);
     }
 
     if (workItem.getSourceOperationTask() != null) {

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/worker/FileBasedReader.java

@@ -19,7 +19,7 @@
 import static com.google.api.client.util.Preconditions.checkNotNull;
 import static com.google.cloud.dataflow.sdk.runners.worker.SourceTranslationUtils.cloudPositionToReaderPosition;
 import static com.google.cloud.dataflow.sdk.runners.worker.SourceTranslationUtils.cloudProgressToReaderProgress;
-import static com.google.cloud.dataflow.sdk.runners.worker.SourceTranslationUtils.forkRequestToApproximateProgress;
+import static com.google.cloud.dataflow.sdk.runners.worker.SourceTranslationUtils.splitRequestToApproximateProgress;
 
 import com.google.api.services.dataflow.model.ApproximateProgress;
 import com.google.cloud.dataflow.sdk.coders.Coder;
@@ -196,38 +196,40 @@ public Progress getProgress() {
     }
 
     @Override
-    public ForkResult requestFork(ForkRequest forkRequest) {
-      checkNotNull(forkRequest);
-
-      // Currently, file-based Reader only supports fork at a byte offset.
-      ApproximateProgress forkProgress = forkRequestToApproximateProgress(forkRequest);
-      com.google.api.services.dataflow.model.Position forkPosition = forkProgress.getPosition();
-      if (forkPosition == null) {
-        LOG.warn("FileBasedReader only supports fork at a Position. Requested: {}", forkRequest);
+    public DynamicSplitResult requestDynamicSplit(DynamicSplitRequest splitRequest) {
+      checkNotNull(splitRequest);
+
+      // Currently, file-based Reader only supports split at a byte offset.
+      ApproximateProgress splitProgress = splitRequestToApproximateProgress(splitRequest);
+      com.google.api.services.dataflow.model.Position splitPosition = splitProgress.getPosition();
+      if (splitPosition == null) {
+        LOG.warn("FileBasedReader only supports split at a Position. Requested: {}",
+            splitRequest);
         return null;
       }
-      Long forkOffset = forkPosition.getByteOffset();
-      if (forkOffset == null) {
-        LOG.warn("FileBasedReader only supports fork at byte offset. Requested: {}", forkPosition);
+      Long splitOffset = splitPosition.getByteOffset();
+      if (splitOffset == null) {
+        LOG.warn("FileBasedReader only supports split at byte offset. Requested: {}",
+            splitPosition);
         return null;
       }
-      if (forkOffset <= offset) {
-        LOG.info("Already progressed to offset {} which is after the requested fork offset {}",
-            offset, forkOffset);
+      if (splitOffset <= offset) {
+        LOG.info("Already progressed to offset {} which is after the requested split offset {}",
+            offset, splitOffset);
         return null;
       }
 
-      if (endOffset != null && forkOffset >= endOffset) {
+      if (endOffset != null && splitOffset >= endOffset) {
         LOG.info(
-            "Fork requested at an offset beyond the end of the current range: {} >= {}",
-            forkOffset, endOffset);
+            "Split requested at an offset beyond the end of the current range: {} >= {}",
+            splitOffset, endOffset);
         return null;
       }
 
-      this.endOffset = forkOffset;
-      LOG.info("Forked FileBasedReader at offset {}", forkOffset);
+      this.endOffset = splitOffset;
+      LOG.info("Split FileBasedReader at offset {}", splitOffset);
 
-      return new ForkResultWithPosition(cloudPositionToReaderPosition(forkPosition));
+      return new DynamicSplitResultWithPosition(cloudPositionToReaderPosition(splitPosition));
     }
 
     /**

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/worker/GroupingShuffleReader.java

@@ -19,7 +19,7 @@
 import static com.google.api.client.util.Preconditions.checkNotNull;
 import static com.google.cloud.dataflow.sdk.runners.worker.SourceTranslationUtils.cloudPositionToReaderPosition;
 import static com.google.cloud.dataflow.sdk.runners.worker.SourceTranslationUtils.cloudProgressToReaderProgress;
-import static com.google.cloud.dataflow.sdk.runners.worker.SourceTranslationUtils.forkRequestToApproximateProgress;
+import static com.google.cloud.dataflow.sdk.runners.worker.SourceTranslationUtils.splitRequestToApproximateProgress;
 
 import com.google.api.client.util.Preconditions;
 import com.google.api.services.dataflow.model.ApproximateProgress;
@@ -225,42 +225,43 @@ public Progress getProgress() {
      * {@code KV<K, Reiterable<V>>} to be returned by the {@link GroupingShuffleReaderIterator}.
      */
     @Override
-    public ForkResult requestFork(ForkRequest forkRequest) {
-      checkNotNull(forkRequest);
-      ApproximateProgress forkProgress = forkRequestToApproximateProgress(forkRequest);
-      com.google.api.services.dataflow.model.Position forkPosition = forkProgress.getPosition();
-      if (forkPosition == null) {
-        LOG.warn("GroupingShuffleReader only supports fork at a Position. Requested: {}",
-            forkRequest);
+    public DynamicSplitResult requestDynamicSplit(DynamicSplitRequest splitRequest) {
+      checkNotNull(splitRequest);
+      ApproximateProgress splitProgress = splitRequestToApproximateProgress(
+          splitRequest);
+      com.google.api.services.dataflow.model.Position splitPosition = splitProgress.getPosition();
+      if (splitPosition == null) {
+        LOG.warn("GroupingShuffleReader only supports split at a Position. Requested: {}",
+            splitRequest);
         return null;
       }
-      String forkShufflePosition = forkPosition.getShufflePosition();
-      if (forkShufflePosition == null) {
-        LOG.warn("GroupingShuffleReader only supports fork at a shuffle position. Requested: {}",
-            forkPosition);
+      String splitShufflePosition = splitPosition.getShufflePosition();
+      if (splitShufflePosition == null) {
+        LOG.warn("GroupingShuffleReader only supports split at a shuffle position. Requested: {}",
+            splitPosition);
         return null;
       }
       ByteArrayShufflePosition newStopPosition =
-          ByteArrayShufflePosition.fromBase64(forkShufflePosition);
+          ByteArrayShufflePosition.fromBase64(splitShufflePosition);
       if (newStopPosition.compareTo(promisedPosition) <= 0) {
         LOG.info("Already progressed to promised shuffle position {} "
-            + "which is after the requested fork shuffle position {}",
-            promisedPosition.encodeBase64(), forkShufflePosition);
+            + "which is after the requested split shuffle position {}",
+            promisedPosition.encodeBase64(), splitShufflePosition);
         return null;
       }
 
       if (this.stopPosition != null && newStopPosition.compareTo(this.stopPosition) >= 0) {
         LOG.info(
-            "Fork requested at a shuffle position beyond the end of the current range: "
+            "Split requested at a shuffle position beyond the end of the current range: "
             + "{} >= current stop position: {}",
-            forkShufflePosition, this.stopPosition.encodeBase64());
+            splitShufflePosition, this.stopPosition.encodeBase64());
         return null;
       }
 
       this.stopPosition = newStopPosition;
-      LOG.info("Forked GroupingShuffleReader at {}", forkShufflePosition);
+      LOG.info("Split GroupingShuffleReader at {}", splitShufflePosition);
 
-      return new ForkResultWithPosition(cloudPositionToReaderPosition(forkPosition));
+      return new DynamicSplitResultWithPosition(cloudPositionToReaderPosition(splitPosition));
     }
 
     /**

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/worker/InMemoryReader.java

@@ -19,7 +19,7 @@
 import static com.google.api.client.util.Preconditions.checkNotNull;
 import static com.google.cloud.dataflow.sdk.runners.worker.SourceTranslationUtils.cloudPositionToReaderPosition;
 import static com.google.cloud.dataflow.sdk.runners.worker.SourceTranslationUtils.cloudProgressToReaderProgress;
-import static com.google.cloud.dataflow.sdk.runners.worker.SourceTranslationUtils.forkRequestToApproximateProgress;
+import static com.google.cloud.dataflow.sdk.runners.worker.SourceTranslationUtils.splitRequestToApproximateProgress;
 import static java.lang.Math.min;
 
 import com.google.api.services.dataflow.model.ApproximateProgress;
@@ -124,38 +124,39 @@ public Progress getProgress() {
     }
 
     @Override
-    public ForkResult requestFork(ForkRequest forkRequest) {
-      checkNotNull(forkRequest);
-
-      com.google.api.services.dataflow.model.Position forkPosition =
-          forkRequestToApproximateProgress(forkRequest).getPosition();
-      if (forkPosition == null) {
-        LOG.warn("InMemoryReader only supports fork at a Position. Requested: {}", forkRequest);
+    public DynamicSplitResult requestDynamicSplit(DynamicSplitRequest splitRequest) {
+      checkNotNull(splitRequest);
+
+      com.google.api.services.dataflow.model.Position splitPosition =
+          splitRequestToApproximateProgress(splitRequest).getPosition();
+      if (splitPosition == null) {
+        LOG.warn("InMemoryReader only supports split at a Position. Requested: {}",
+            splitRequest);
         return null;
       }
 
-      Long forkIndex = forkPosition.getRecordIndex();
-      if (forkIndex == null) {
-        LOG.warn("InMemoryReader only supports fork at a record index. Requested: {}",
-            forkPosition);
+      Long splitIndex = splitPosition.getRecordIndex();
+      if (splitIndex == null) {
+        LOG.warn("InMemoryReader only supports split at a record index. Requested: {}",
+            splitPosition);
         return null;
       }
-      if (forkIndex <= index) {
-        LOG.info("Already progressed to index {} which is after the requested fork index {}",
-            index, forkIndex);
+      if (splitIndex <= index) {
+        LOG.info("Already progressed to index {} which is after the requested split index {}",
+            index, splitIndex);
         return null;
       }
-      if (forkIndex >= endPosition) {
+      if (splitIndex >= endPosition) {
         LOG.info(
-            "Fork requested at an index beyond the end of the current range: {} >= {}",
-            forkIndex, endPosition);
+            "Split requested at an index beyond the end of the current range: {} >= {}",
+            splitIndex, endPosition);
         return null;
       }
 
-      this.endPosition = forkIndex.intValue();
-      LOG.info("Forked InMemoryReader at index {}", forkIndex);
+      this.endPosition = splitIndex.intValue();
+      LOG.info("Split InMemoryReader at index {}", splitIndex);
 
-      return new ForkResultWithPosition(cloudPositionToReaderPosition(forkPosition));
+      return new DynamicSplitResultWithPosition(cloudPositionToReaderPosition(splitPosition));
     }
   }
 }

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/worker/SourceTranslationUtils.java

@@ -85,14 +85,16 @@ public static Source sourceSpecToCloudSource(@Nullable SourceFormat.SourceSpec s
     return (spec == null) ? null : ((DataflowSourceSpec) spec).cloudSource;
   }
 
-  public static ApproximateProgress forkRequestToApproximateProgress(
-      @Nullable Reader.ForkRequest stopRequest) {
-    return (stopRequest == null) ? null : ((DataflowForkRequest) stopRequest).approximateProgress;
+  public static ApproximateProgress splitRequestToApproximateProgress(
+      @Nullable Reader.DynamicSplitRequest splitRequest) {
+    return (splitRequest == null)
+        ? null : ((DataflowDynamicSplitRequest) splitRequest).approximateProgress;
   }
 
-  public static Reader.ForkRequest toForkRequest(
+  public static Reader.DynamicSplitRequest toDynamicSplitRequest(
       @Nullable ApproximateProgress approximateProgress) {
-    return (approximateProgress == null) ? null : new DataflowForkRequest(approximateProgress);
+    return (approximateProgress == null)
+        ? null : new DataflowDynamicSplitRequest(approximateProgress);
   }
 
   static class DataflowReaderProgress implements Reader.Progress {
@@ -175,10 +177,10 @@ public static Source dictionaryToCloudSource(Map<String, Object> params) throws
     return res;
   }
 
-  private static class DataflowForkRequest implements Reader.ForkRequest {
+  private static class DataflowDynamicSplitRequest implements Reader.DynamicSplitRequest {
     public final ApproximateProgress approximateProgress;
 
-    private DataflowForkRequest(ApproximateProgress approximateProgress) {
+    private DataflowDynamicSplitRequest(ApproximateProgress approximateProgress) {
       this.approximateProgress = approximateProgress;
     }
   }

sdk/src/main/java/com/google/cloud/dataflow/sdk/util/common/worker/MapTaskExecutor.java

@@ -85,8 +85,9 @@ public Reader.Progress getWorkerProgress() throws Exception {
   }
 
   @Override
-  public Reader.ForkResult requestFork(Reader.ForkRequest forkRequest) throws Exception {
-    return getReadOperation().requestFork(forkRequest);
+  public Reader.DynamicSplitResult requestDynamicSplit(
+      Reader.DynamicSplitRequest splitRequest) throws Exception {
+    return getReadOperation().requestDynamicSplit(splitRequest);
   }
 
   ReadOperation getReadOperation() throws Exception {

sdk/src/main/java/com/google/cloud/dataflow/sdk/util/common/worker/ReadOperation.java

@@ -220,20 +220,21 @@ public Reader.Progress getProgress() {
   }
 
   /**
-   * Relays the fork request to {@code ReaderIterator}.
+   * Relays the split request to {@code ReaderIterator}.
    */
-  public Reader.ForkResult requestFork(Reader.ForkRequest forkRequest) {
+  public Reader.DynamicSplitResult requestDynamicSplit(Reader.DynamicSplitRequest splitRequest) {
     synchronized (initializationStateLock) {
       if (isFinished()) {
         LOG.warn("Iterator is in the Finished state, returning null stop position.");
         return null;
       }
       synchronized (sourceIteratorLock) {
         if (readerIterator == null) {
-          LOG.warn("Iterator has not been initialized, refusing to fork at {}", forkRequest);
+          LOG.warn("Iterator has not been initialized, refusing to split at {}",
+              splitRequest);
           return null;
         }
-        return readerIterator.requestFork(forkRequest);
+        return readerIterator.requestDynamicSplit(splitRequest);
       }
     }
   }