beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [1/9] beam git commit: Reify delay and alignment in AfterProcessingTime transform
Date Wed, 15 Feb 2017 04:51:43 GMT
Repository: beam
Updated Branches:
  refs/heads/master 9061c65e3 -> db19c7df5


Reify delay and alignment in AfterProcessingTime transform


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/40c4a5cb
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/40c4a5cb
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/40c4a5cb

Branch: refs/heads/master
Commit: 40c4a5cb6eaa0350a26fe1f215eb812541a7b105
Parents: bea101a
Author: Kenneth Knowles <klk@google.com>
Authored: Sun Feb 12 15:03:48 2017 -0800
Committer: Kenneth Knowles <klk@google.com>
Committed: Tue Feb 14 14:55:48 2017 -0800

----------------------------------------------------------------------
 .../AfterDelayFromFirstElementStateMachine.java |   4 +-
 .../AfterProcessingTimeStateMachine.java        |   2 +
 ...rSynchronizedProcessingTimeStateMachine.java |   1 +
 .../core/triggers/TriggerStateMachines.java     |  36 ++-
 .../core/triggers/TriggerStateMachinesTest.java |   7 +-
 .../windowing/AfterDelayFromFirstElement.java   | 240 -------------------
 .../windowing/AfterProcessingTime.java          | 105 ++++++--
 .../AfterSynchronizedProcessingTime.java        |  25 +-
 .../windowing/TimestampTransform.java           |  64 +++++
 .../windowing/AfterProcessingTimeTest.java      |   2 +-
 10 files changed, 192 insertions(+), 294 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/40c4a5cb/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterDelayFromFirstElementStateMachine.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterDelayFromFirstElementStateMachine.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterDelayFromFirstElementStateMachine.java
index b720644..4444c22 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterDelayFromFirstElementStateMachine.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterDelayFromFirstElementStateMachine.java
@@ -47,6 +47,8 @@ import org.joda.time.format.PeriodFormatter;
  *
  * <p>This class is for internal use only and may change at any time.
  */
+// This class should be inlined to subclasses and deleted, simplifying them too
+// https://issues.apache.org/jira/browse/BEAM-1486
 @Experimental(Experimental.Kind.TRIGGER)
 public abstract class AfterDelayFromFirstElementStateMachine extends OnceTriggerStateMachine
{
 
@@ -250,7 +252,7 @@ public abstract class AfterDelayFromFirstElementStateMachine extends OnceTrigger
   /**
    * A {@link SerializableFunction} to delay the timestamp at which this triggers fires.
    */
-  private static final class DelayFn implements SerializableFunction<Instant, Instant>
{
+  static final class DelayFn implements SerializableFunction<Instant, Instant> {
     private final Duration delay;
 
     public DelayFn(Duration delay) {

http://git-wip-us.apache.org/repos/asf/beam/blob/40c4a5cb/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterProcessingTimeStateMachine.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterProcessingTimeStateMachine.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterProcessingTimeStateMachine.java
index 2490463..eaf5613 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterProcessingTimeStateMachine.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterProcessingTimeStateMachine.java
@@ -34,6 +34,8 @@ import org.joda.time.Instant;
  * AfterDelayFromFirstElementStateMachine#plusDelayOf} or {@link
  * AfterDelayFromFirstElementStateMachine#alignedTo}.
  */
+// The superclass should be inlined here, its only real use
+// https://issues.apache.org/jira/browse/BEAM-1486
 @Experimental(Experimental.Kind.TRIGGER)
 public class AfterProcessingTimeStateMachine extends AfterDelayFromFirstElementStateMachine
{
 

http://git-wip-us.apache.org/repos/asf/beam/blob/40c4a5cb/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterSynchronizedProcessingTimeStateMachine.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterSynchronizedProcessingTimeStateMachine.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterSynchronizedProcessingTimeStateMachine.java
index 000f6e7..1319a13 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterSynchronizedProcessingTimeStateMachine.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterSynchronizedProcessingTimeStateMachine.java
@@ -25,6 +25,7 @@ import org.apache.beam.sdk.transforms.SerializableFunction;
 import org.apache.beam.sdk.util.TimeDomain;
 import org.joda.time.Instant;
 
+// This should not really have the superclass https://issues.apache.org/jira/browse/BEAM-1486
 class AfterSynchronizedProcessingTimeStateMachine extends AfterDelayFromFirstElementStateMachine
{
 
   @Override

http://git-wip-us.apache.org/repos/asf/beam/blob/40c4a5cb/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachines.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachines.java
b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachines.java
index 1be7981..f0e9d21 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachines.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachines.java
@@ -18,14 +18,16 @@
 package org.apache.beam.runners.core.triggers;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
+import com.google.common.collect.FluentIterable;
 import com.google.common.collect.Lists;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.List;
+import javax.annotation.Nonnull;
 import org.apache.beam.runners.core.triggers.TriggerStateMachine.OnceTriggerStateMachine;
 import org.apache.beam.sdk.transforms.SerializableFunction;
 import org.apache.beam.sdk.transforms.windowing.AfterAll;
-import org.apache.beam.sdk.transforms.windowing.AfterDelayFromFirstElement;
 import org.apache.beam.sdk.transforms.windowing.AfterEach;
 import org.apache.beam.sdk.transforms.windowing.AfterFirst;
 import org.apache.beam.sdk.transforms.windowing.AfterPane;
@@ -36,6 +38,7 @@ import org.apache.beam.sdk.transforms.windowing.DefaultTrigger;
 import org.apache.beam.sdk.transforms.windowing.Never.NeverTrigger;
 import org.apache.beam.sdk.transforms.windowing.OrFinallyTrigger;
 import org.apache.beam.sdk.transforms.windowing.Repeatedly;
+import org.apache.beam.sdk.transforms.windowing.TimestampTransform;
 import org.apache.beam.sdk.transforms.windowing.Trigger;
 import org.apache.beam.sdk.transforms.windowing.Trigger.OnceTrigger;
 import org.apache.beam.sdk.util.ReshuffleTrigger;
@@ -171,18 +174,37 @@ public class TriggerStateMachines {
     }
 
     private OnceTriggerStateMachine evaluateSpecific(AfterProcessingTime v) {
-      return evaluateSpecific((AfterDelayFromFirstElement) v);
-    }
-
-    private OnceTriggerStateMachine evaluateSpecific(final AfterDelayFromFirstElement v)
{
       return new AfterDelayFromFirstElementStateMachineAdapter(v);
     }
 
     private static class AfterDelayFromFirstElementStateMachineAdapter
         extends AfterDelayFromFirstElementStateMachine {
 
-      public AfterDelayFromFirstElementStateMachineAdapter(AfterDelayFromFirstElement v)
{
-        this(v.getTimeDomain(), v.getTimestampMappers());
+      private static final Function<TimestampTransform, SerializableFunction<Instant,
Instant>>
+          CONVERT_TIMESTAMP_TRANSFORM =
+              new Function<TimestampTransform, SerializableFunction<Instant, Instant>>()
{
+                @Override
+                public SerializableFunction<Instant, Instant> apply(
+                    @Nonnull TimestampTransform transform) {
+                  if (transform instanceof TimestampTransform.Delay) {
+                    return new DelayFn(((TimestampTransform.Delay) transform).getDelay());
+                  } else if (transform instanceof TimestampTransform.AlignTo) {
+                    TimestampTransform.AlignTo alignTo = (TimestampTransform.AlignTo) transform;
+                    return new AlignFn(alignTo.getPeriod(), alignTo.getOffset());
+                  } else {
+                    throw new IllegalArgumentException(
+                        String.format(
+                            "Unknown %s: %s", TimestampTransform.class.getSimpleName(), transform));
+                  }
+                }
+              };
+
+      public AfterDelayFromFirstElementStateMachineAdapter(AfterProcessingTime v) {
+        this(
+            TimeDomain.PROCESSING_TIME,
+            FluentIterable.from(v.getTimestampTransforms())
+                .transform(CONVERT_TIMESTAMP_TRANSFORM)
+                .toList());
       }
 
       private AfterDelayFromFirstElementStateMachineAdapter(

http://git-wip-us.apache.org/repos/asf/beam/blob/40c4a5cb/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/TriggerStateMachinesTest.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/TriggerStateMachinesTest.java
b/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/TriggerStateMachinesTest.java
index 37f8f10..26c0597 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/TriggerStateMachinesTest.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/triggers/TriggerStateMachinesTest.java
@@ -23,7 +23,6 @@ import static org.junit.Assert.assertThat;
 
 import org.apache.beam.runners.core.triggers.TriggerStateMachine.OnceTriggerStateMachine;
 import org.apache.beam.sdk.transforms.windowing.AfterAll;
-import org.apache.beam.sdk.transforms.windowing.AfterDelayFromFirstElement;
 import org.apache.beam.sdk.transforms.windowing.AfterEach;
 import org.apache.beam.sdk.transforms.windowing.AfterFirst;
 import org.apache.beam.sdk.transforms.windowing.AfterPane;
@@ -64,7 +63,7 @@ public class TriggerStateMachinesTest {
     Duration minutes = Duration.standardMinutes(94);
     Duration hours = Duration.standardHours(13);
 
-    AfterDelayFromFirstElement trigger =
+    AfterProcessingTime trigger =
         AfterProcessingTime.pastFirstElementInPane().plusDelayOf(minutes).alignedTo(hours);
 
     AfterDelayFromFirstElementStateMachine machine =
@@ -72,10 +71,6 @@ public class TriggerStateMachinesTest {
             TriggerStateMachines.stateMachineForOnceTrigger(trigger);
 
     assertThat(machine.getTimeDomain(), equalTo(TimeDomain.PROCESSING_TIME));
-
-    // This equality is function equality, but due to the structure of the code (no serialization)
-    // it is OK to check
-    assertThat(machine.getTimestampMappers(), equalTo(trigger.getTimestampMappers()));
   }
 
   @Test

http://git-wip-us.apache.org/repos/asf/beam/blob/40c4a5cb/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterDelayFromFirstElement.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterDelayFromFirstElement.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterDelayFromFirstElement.java
deleted file mode 100644
index da6d74a..0000000
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterDelayFromFirstElement.java
+++ /dev/null
@@ -1,240 +0,0 @@
-/*
- * 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.beam.sdk.transforms.windowing;
-
-import com.google.common.collect.ImmutableList;
-import java.util.List;
-import java.util.Locale;
-import java.util.Objects;
-import org.apache.beam.sdk.annotations.Experimental;
-import org.apache.beam.sdk.transforms.SerializableFunction;
-import org.apache.beam.sdk.transforms.windowing.Trigger.OnceTrigger;
-import org.apache.beam.sdk.util.TimeDomain;
-import org.joda.time.Duration;
-import org.joda.time.Instant;
-import org.joda.time.format.PeriodFormat;
-import org.joda.time.format.PeriodFormatter;
-
-/**
- * A base class for triggers that happen after a processing time delay from the arrival
- * of the first element in a pane.
- *
- * <p>This class is for internal use only and may change at any time.
- */
-@Experimental(Experimental.Kind.TRIGGER)
-public abstract class AfterDelayFromFirstElement extends OnceTrigger {
-
-  protected static final List<SerializableFunction<Instant, Instant>> IDENTITY
=
-      ImmutableList.<SerializableFunction<Instant, Instant>>of();
-
-  private static final PeriodFormatter PERIOD_FORMATTER = PeriodFormat.wordBased(Locale.ENGLISH);
-
-  /**
-   * To complete an implementation, return a new instance like this one, but incorporating
-   * the provided timestamp mapping functions. Generally should be used by calling the
-   * constructor of this class from the constructor of the subclass.
-   */
-  protected abstract AfterDelayFromFirstElement newWith(
-      List<SerializableFunction<Instant, Instant>> transform);
-
-  /**
-   * A list of timestampMappers m1, m2, m3, ... m_n considered to be composed in sequence.
The
-   * overall mapping for an instance `instance` is `m_n(... m3(m2(m1(instant))`,
-   * implemented via #computeTargetTimestamp
-   */
-  protected final List<SerializableFunction<Instant, Instant>> timestampMappers;
-
-  private final TimeDomain timeDomain;
-
-  public AfterDelayFromFirstElement(
-      TimeDomain timeDomain,
-      List<SerializableFunction<Instant, Instant>> timestampMappers) {
-    super(null);
-    this.timestampMappers = timestampMappers;
-    this.timeDomain = timeDomain;
-  }
-
-  /**
-   * The time domain according for which this trigger sets timers.
-   */
-  public TimeDomain getTimeDomain() {
-    return timeDomain;
-  }
-
-  /**
-   * The mapping functions applied to the arrival time of an element to determine when to
-   * set a wake-up timer for triggering.
-   */
-  public List<SerializableFunction<Instant, Instant>> getTimestampMappers() {
-    return timestampMappers;
-  }
-
-  /**
-   * Aligns timestamps to the smallest multiple of {@code size} since the {@code offset}
greater
-   * than the timestamp.
-   *
-   * <p>TODO: Consider sharing this with FixedWindows, and bring over the equivalent
of
-   * CalendarWindows.
-   */
-  public AfterDelayFromFirstElement alignedTo(final Duration size, final Instant offset)
{
-    return newWith(new AlignFn(size, offset));
-  }
-
-  /**
-   * Aligns the time to be the smallest multiple of {@code size} greater than the timestamp
-   * since the epoch.
-   */
-  public AfterDelayFromFirstElement alignedTo(final Duration size) {
-    return alignedTo(size, new Instant(0));
-  }
-
-  /**
-   * Adds some delay to the original target time.
-   *
-   * @param delay the delay to add
-   * @return An updated time trigger that will wait the additional time before firing.
-   */
-  public AfterDelayFromFirstElement plusDelayOf(final Duration delay) {
-    return newWith(new DelayFn(delay));
-  }
-
-  /**
-   * @deprecated This will be removed in the next major version. Please use only
-   *             {@link #plusDelayOf} and {@link #alignedTo}.
-   */
-  @Deprecated
-  public OnceTrigger mappedTo(SerializableFunction<Instant, Instant> timestampMapper)
{
-    return newWith(timestampMapper);
-  }
-
-  @Override
-  public boolean isCompatible(Trigger other) {
-    if (!getClass().equals(other.getClass())) {
-      return false;
-    }
-
-    AfterDelayFromFirstElement that = (AfterDelayFromFirstElement) other;
-    return this.timestampMappers.equals(that.timestampMappers);
-  }
-
-
-  private AfterDelayFromFirstElement newWith(
-      SerializableFunction<Instant, Instant> timestampMapper) {
-    return newWith(
-        ImmutableList.<SerializableFunction<Instant, Instant>>builder()
-            .addAll(timestampMappers)
-            .add(timestampMapper)
-            .build());
-  }
-
-  @Override
-  public Instant getWatermarkThatGuaranteesFiring(BoundedWindow window) {
-    return BoundedWindow.TIMESTAMP_MAX_VALUE;
-  }
-
-  protected Instant computeTargetTimestamp(Instant time) {
-    Instant result = time;
-    for (SerializableFunction<Instant, Instant> timestampMapper : timestampMappers)
{
-      result = timestampMapper.apply(result);
-    }
-    return result;
-  }
-
-  /**
-   * A {@link SerializableFunction} to delay the timestamp at which this triggers fires.
-   */
-  private static final class DelayFn implements SerializableFunction<Instant, Instant>
{
-    private final Duration delay;
-
-    public DelayFn(Duration delay) {
-      this.delay = delay;
-    }
-
-    @Override
-    public Instant apply(Instant input) {
-      return input.plus(delay);
-    }
-
-    @Override
-    public boolean equals(Object object) {
-      if (object == this) {
-        return true;
-      }
-
-      if (!(object instanceof DelayFn)) {
-        return false;
-      }
-
-      return this.delay.equals(((DelayFn) object).delay);
-    }
-
-    @Override
-    public int hashCode() {
-      return Objects.hash(delay);
-    }
-
-    @Override
-    public String toString() {
-      return PERIOD_FORMATTER.print(delay.toPeriod());
-    }
-  }
-
-  /**
-   * A {@link SerializableFunction} to align an instant to the nearest interval boundary.
-   */
-  static final class AlignFn implements SerializableFunction<Instant, Instant> {
-    private final Duration size;
-    private final Instant offset;
-
-
-    /**
-     * Aligns timestamps to the smallest multiple of {@code size} since the {@code offset}
greater
-     * than the timestamp.
-     */
-    public AlignFn(Duration size, Instant offset) {
-      this.size = size;
-      this.offset = offset;
-    }
-
-    @Override
-    public Instant apply(Instant point) {
-      long millisSinceStart = new Duration(offset, point).getMillis() % size.getMillis();
-      return millisSinceStart == 0 ? point : point.plus(size).minus(millisSinceStart);
-    }
-
-    @Override
-    public boolean equals(Object object) {
-      if (object == this) {
-        return true;
-      }
-
-      if (!(object instanceof AlignFn)) {
-        return false;
-      }
-
-      AlignFn other = (AlignFn) object;
-      return other.size.equals(this.size)
-          && other.offset.equals(this.offset);
-    }
-
-    @Override
-    public int hashCode() {
-      return Objects.hash(size, offset);
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/beam/blob/40c4a5cb/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTime.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTime.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTime.java
index 09f288e..eda269a 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTime.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTime.java
@@ -17,40 +17,87 @@
  */
 package org.apache.beam.sdk.transforms.windowing;
 
+import com.google.common.collect.ImmutableList;
+import java.util.Collections;
 import java.util.List;
+import java.util.Locale;
 import java.util.Objects;
 import org.apache.beam.sdk.annotations.Experimental;
-import org.apache.beam.sdk.transforms.SerializableFunction;
-import org.apache.beam.sdk.util.TimeDomain;
+import org.apache.beam.sdk.transforms.windowing.Trigger.OnceTrigger;
+import org.joda.time.Duration;
 import org.joda.time.Instant;
+import org.joda.time.format.PeriodFormat;
+import org.joda.time.format.PeriodFormatter;
 
 /**
  * {@code AfterProcessingTime} triggers fire based on the current processing time. They operate
in
  * the real-time domain.
- *
- * <p>The time at which to fire the timer can be adjusted via the methods in
- * {@link AfterDelayFromFirstElement}, such as {@link AfterDelayFromFirstElement#plusDelayOf}
or
- * {@link AfterDelayFromFirstElement#alignedTo}.
  */
 @Experimental(Experimental.Kind.TRIGGER)
-public class AfterProcessingTime extends AfterDelayFromFirstElement {
+public class AfterProcessingTime extends OnceTrigger {
+
+  private static final PeriodFormatter DURATION_FORMATTER = PeriodFormat.wordBased(Locale.ENGLISH);
+
+  private final List<TimestampTransform> timestampTransforms;
 
-  private AfterProcessingTime(List<SerializableFunction<Instant, Instant>> transforms)
{
-    super(TimeDomain.PROCESSING_TIME, transforms);
+  public AfterProcessingTime(List<TimestampTransform> timestampTransforms) {
+    super(null);
+    this.timestampTransforms = timestampTransforms;
   }
 
   /**
-   * Creates a trigger that fires when the current processing time passes the processing
time
-   * at which this trigger saw the first element in a pane.
+   * Creates a trigger that fires when the current processing time passes the processing
time at
+   * which this trigger saw the first element in a pane.
    */
   public static AfterProcessingTime pastFirstElementInPane() {
-    return new AfterProcessingTime(IDENTITY);
+    return new AfterProcessingTime(Collections.<TimestampTransform>emptyList());
+  }
+
+  /**
+   * The transforms applied to the arrival time of an element to determine when this trigger
allows
+   * output.
+   */
+  public List<TimestampTransform> getTimestampTransforms() {
+    return timestampTransforms;
+  }
+
+  /**
+   * Adds some delay to the original target time.
+   *
+   * @param delay the delay to add
+   * @return An updated time trigger that will wait the additional time before firing.
+   */
+  public AfterProcessingTime plusDelayOf(final Duration delay) {
+    return new AfterProcessingTime(
+        ImmutableList.<TimestampTransform>builder()
+            .addAll(timestampTransforms)
+            .add(TimestampTransform.delay(delay))
+            .build());
+  }
+
+  /**
+   * Aligns timestamps to the smallest multiple of {@code period} since the {@code offset}
greater
+   * than the timestamp.
+   */
+  public AfterProcessingTime alignedTo(final Duration period, final Instant offset) {
+    return new AfterProcessingTime(
+        ImmutableList.<TimestampTransform>builder()
+            .addAll(timestampTransforms)
+            .add(TimestampTransform.alignTo(period, offset))
+            .build());
+  }
+
+  /**
+   * Aligns the time to be the smallest multiple of {@code period} greater than the epoch
+   * boundary (aka {@code new Instant(0)}).
+   */
+  public AfterProcessingTime alignedTo(final Duration period) {
+    return alignedTo(period, new Instant(0));
   }
 
   @Override
-  protected AfterProcessingTime newWith(
-      List<SerializableFunction<Instant, Instant>> transforms) {
-    return new AfterProcessingTime(transforms);
+  public boolean isCompatible(Trigger other) {
+    return this.equals(other);
   }
 
   @Override
@@ -60,17 +107,28 @@ public class AfterProcessingTime extends AfterDelayFromFirstElement {
 
   @Override
   protected Trigger getContinuationTrigger(List<Trigger> continuationTriggers) {
-    return new AfterSynchronizedProcessingTime();
+    return AfterSynchronizedProcessingTime.ofFirstElement();
   }
 
   @Override
   public String toString() {
     StringBuilder builder = new StringBuilder("AfterProcessingTime.pastFirstElementInPane()");
-    for (SerializableFunction<Instant, Instant> delayFn : timestampMappers) {
-      builder
-          .append(".plusDelayOf(")
-          .append(delayFn)
-          .append(")");
+    for (TimestampTransform transform : getTimestampTransforms()) {
+      if (transform instanceof TimestampTransform.Delay) {
+        TimestampTransform.Delay delay = (TimestampTransform.Delay) transform;
+        builder
+            .append(".plusDelayOf(")
+            .append(DURATION_FORMATTER.print(delay.getDelay().toPeriod()))
+            .append(")");
+      } else if (transform instanceof TimestampTransform.AlignTo) {
+        TimestampTransform.AlignTo alignTo = (TimestampTransform.AlignTo) transform;
+        builder
+            .append(".alignedTo(")
+            .append(DURATION_FORMATTER.print(alignTo.getPeriod().toPeriod()))
+            .append(", ")
+            .append(alignTo.getOffset())
+            .append(")");
+      }
     }
 
     return builder.toString();
@@ -84,12 +142,13 @@ public class AfterProcessingTime extends AfterDelayFromFirstElement {
     if (!(obj instanceof AfterProcessingTime)) {
       return false;
     }
+
     AfterProcessingTime that = (AfterProcessingTime) obj;
-    return Objects.equals(this.timestampMappers, that.timestampMappers);
+    return getTimestampTransforms().equals(that.getTimestampTransforms());
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(getClass(), this.timestampMappers);
+    return Objects.hash(getTimestampTransforms());
   }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/40c4a5cb/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterSynchronizedProcessingTime.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterSynchronizedProcessingTime.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterSynchronizedProcessingTime.java
index b6258f8..6249954 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterSynchronizedProcessingTime.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/AfterSynchronizedProcessingTime.java
@@ -18,21 +18,22 @@
 package org.apache.beam.sdk.transforms.windowing;
 
 import com.google.common.base.Objects;
-import java.util.Collections;
 import java.util.List;
-import org.apache.beam.sdk.transforms.SerializableFunction;
-import org.apache.beam.sdk.util.TimeDomain;
+import org.apache.beam.sdk.transforms.windowing.Trigger.OnceTrigger;
 import org.joda.time.Instant;
 
 /**
- * A trigger that fires after synchronized processing time has reached a shared
- * threshold between upstream workers.
+ * A trigger that fires after synchronized processing time has reached the processing time
of the
+ * first element's arrival.
+ *
+ * <p>This is for internal, primarily as a "continuation trigger" for {@link AfterProcessingTime}
+ * triggers. In that use, this trigger is ready as soon as all upstream workers processing
time
+ * clocks have caught up to the moment that input arrived.
  */
-public class AfterSynchronizedProcessingTime extends AfterDelayFromFirstElement {
+public class AfterSynchronizedProcessingTime extends OnceTrigger {
 
   public AfterSynchronizedProcessingTime() {
-    super(TimeDomain.SYNCHRONIZED_PROCESSING_TIME,
-        Collections.<SerializableFunction<Instant, Instant>>emptyList());
+    super(null);
   }
 
   @Override
@@ -59,12 +60,4 @@ public class AfterSynchronizedProcessingTime extends AfterDelayFromFirstElement
   public int hashCode() {
     return Objects.hashCode(AfterSynchronizedProcessingTime.class);
   }
-
-  @Override
-  protected AfterSynchronizedProcessingTime
-      newWith(List<SerializableFunction<Instant, Instant>> transforms) {
-    // ignore transforms
-    return this;
-  }
-
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/40c4a5cb/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/TimestampTransform.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/TimestampTransform.java
b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/TimestampTransform.java
new file mode 100644
index 0000000..b16e968
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/TimestampTransform.java
@@ -0,0 +1,64 @@
+/*
+ * 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.beam.sdk.transforms.windowing;
+
+import com.google.auto.value.AutoValue;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+
+/** An abstract description of a standardized transformation on timestamps. */
+public abstract class TimestampTransform {
+
+  /** Returns a transform that shifts a timestamp later by {@code delay}. */
+  public static TimestampTransform delay(Duration delay) {
+    return new AutoValue_TimestampTransform_Delay(delay);
+  }
+
+  /**
+   * Returns a transform that aligns a timestamp to the next boundary of {@code period},
starting
+   * from {@code offset}.
+   */
+  public static TimestampTransform alignTo(Duration period, Instant offset) {
+    return new AutoValue_TimestampTransform_AlignTo(period, offset);
+  }
+
+  /**
+   * Returns a transform that aligns a timestamp to the next boundary of {@code period},
starting
+   * from the start of the epoch.
+   */
+  public static TimestampTransform alignTo(Duration period) {
+    return alignTo(period, new Instant(0));
+  }
+
+  /**
+   * Represents the transform that aligns a timestamp to the next boundary of {@link #getPeriod()}
+   * start at {@link #getOffset()}.
+   */
+  @AutoValue
+  public abstract static class AlignTo extends TimestampTransform {
+    public abstract Duration getPeriod();
+
+    public abstract Instant getOffset();
+  }
+
+  /** Represents the transform that delays a timestamp by {@link #getDelay()}. */
+  @AutoValue
+  public abstract static class Delay extends TimestampTransform {
+    public abstract Duration getDelay();
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/40c4a5cb/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTimeTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTimeTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTimeTest.java
index 4984d7c..006ff00 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTimeTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/windowing/AfterProcessingTimeTest.java
@@ -45,7 +45,7 @@ public class AfterProcessingTimeTest {
     OnceTrigger firstElementPlus1 =
         AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.standardHours(1));
     assertEquals(
-        new AfterSynchronizedProcessingTime(),
+        AfterSynchronizedProcessingTime.ofFirstElement(),
         firstElementPlus1.getContinuationTrigger());
   }
 


Mime
View raw message