Merge 7c7f96f into c80ef18

apache · Mar 24, 2017 · 37fc92f · 37fc92f
2 parents c80ef18 + 7c7f96f
commit 37fc92f
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 4 deletions.
diff --git a/...rect-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java b/...rect-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
@@ -41,6 +41,7 @@
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
+import org.apache.beam.sdk.util.CoderUtils;
 import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.values.PBegin;
 import org.apache.beam.sdk.values.PCollection;
@@ -187,6 +188,9 @@ private UnboundedReader<OutputT> getReader(UnboundedSourceShard<OutputT, Checkpo
       UnboundedReader<OutputT> existing = shard.getExistingReader();
       if (existing == null) {
         CheckpointMarkT checkpoint = shard.getCheckpoint();
+        if (checkpoint != null) {
+          checkpoint = CoderUtils.clone(shard.getSource().getCheckpointMarkCoder(), checkpoint);
+        }
         return shard
             .getSource()
             .createReader(evaluationContext.getPipelineOptions(), checkpoint);

diff --git a/...-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java b/...-java/src/test/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactoryTest.java
@@ -458,6 +458,9 @@ public List<? extends UnboundedSource<T, TestCheckpointMark>> generateInitialSpl
     @Override
     public UnboundedSource.UnboundedReader<T> createReader(
         PipelineOptions options, @Nullable TestCheckpointMark checkpointMark) {
+      checkState(
+          checkpointMark == null || checkpointMark.decoded,
+          "Cannot resume from a checkpoint that has not been decoded");
       return new TestUnboundedReader(elems, checkpointMark == null ? -1 : checkpointMark.index);
     }
 
@@ -564,6 +567,7 @@ public void close() throws IOException {
   private static class TestCheckpointMark implements CheckpointMark {
     final int index;
     private boolean finalized = false;
+    private boolean decoded = false;
 
     private TestCheckpointMark(int index) {
       this.index = index;
@@ -573,6 +577,10 @@ private TestCheckpointMark(int index) {
     public void finalizeCheckpoint() throws IOException {
       checkState(
           !finalized, "%s was finalized more than once", TestCheckpointMark.class.getSimpleName());
+      checkState(
+          !decoded,
+          "%s was finalized after being decoded",
+          TestCheckpointMark.class.getSimpleName());
       finalized = true;
     }
 
@@ -594,7 +602,9 @@ public void encode(
       public TestCheckpointMark decode(
           InputStream inStream, org.apache.beam.sdk.coders.Coder.Context context)
           throws IOException {
-        return new TestCheckpointMark(VarInt.decodeInt(inStream));
+        TestCheckpointMark decoded = new TestCheckpointMark(VarInt.decodeInt(inStream));
+        decoded.decoded = true;
+        return decoded;
       }
     }
   }

diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/io/UnboundedSource.java
@@ -76,10 +76,8 @@ public abstract UnboundedReader<OutputT> createReader(
       PipelineOptions options, @Nullable CheckpointMarkT checkpointMark) throws IOException;
 
   /**
-   * Returns a {@link Coder} for encoding and decoding the checkpoints for this source, or
-   * null if the checkpoints do not need to be durably committed.
+   * Returns a {@link Coder} for encoding and decoding the checkpoints for this source.
    */
-  @Nullable
   public abstract Coder<CheckpointMarkT> getCheckpointMarkCoder();
 
   /**