You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2016/10/20 01:02:10 UTC
[1/6] incubator-beam git commit: Tighten up tests for StateId
validation
Repository: incubator-beam
Updated Branches:
refs/heads/master 6c6f824aa -> 307819327
Tighten up tests for StateId validation
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/507cba8f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/507cba8f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/507cba8f
Branch: refs/heads/master
Commit: 507cba8f0df44b1c11c7621f00ff492243bdd1f1
Parents: 4e14ac2
Author: Kenneth Knowles <kl...@google.com>
Authored: Thu Oct 13 23:18:40 2016 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Tue Oct 18 13:08:55 2016 -0700
----------------------------------------------------------------------
.../apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/507cba8f/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java
index 26a7b74..230e6ca 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java
@@ -139,7 +139,7 @@ public class DoFnSignaturesTest {
DoFnSignatures.INSTANCE.getSignature(
new DoFn<String, String>() {
@StateId("foo")
- String bizzle = "bazzle";
+ private final String bizzle = "bazzle";
@ProcessElement
public void foo(ProcessContext context) {}
@@ -162,7 +162,8 @@ public class DoFnSignaturesTest {
StateSpecs.value(VarIntCoder.of());
@StateId("my-state-id")
- StateSpec<Object, ValueState<Long>> myfield2 = StateSpecs.value(VarLongCoder.of());
+ private final StateSpec<Object, ValueState<Long>> myfield2 =
+ StateSpecs.value(VarLongCoder.of());
@ProcessElement
public void foo(ProcessContext context) {}
@@ -245,6 +246,7 @@ public class DoFnSignaturesTest {
Matchers.<TypeDescriptor<?>>equalTo(new TypeDescriptor<ValueState<Integer>>() {}));
}
+
private static class DoFnForTestSimpleStateIdNamedDoFn extends DoFn<KV<String, Integer>, Long> {
@StateId("foo")
private final StateSpec<Object, ValueState<Integer>> bizzle =
[3/6] incubator-beam git commit: Add DoFnSignatures analysis of timer
declarations
Posted by ke...@apache.org.
Add DoFnSignatures analysis of timer declarations
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/ccefc6f2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/ccefc6f2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/ccefc6f2
Branch: refs/heads/master
Commit: ccefc6f22050b0da1471e2b68a0787e9a861639c
Parents: 1b780c2
Author: Kenneth Knowles <kl...@google.com>
Authored: Thu Oct 13 19:53:31 2016 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Wed Oct 19 17:52:21 2016 -0700
----------------------------------------------------------------------
.../sdk/transforms/reflect/DoFnSignature.java | 45 ++++
.../sdk/transforms/reflect/DoFnSignatures.java | 95 +++++++-
.../transforms/reflect/DoFnSignaturesTest.java | 238 +++++++++++++++++++
3 files changed, 376 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ccefc6f2/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
index 5e261a4..e54b361 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java
@@ -30,6 +30,7 @@ import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.DoFn.ProcessContinuation;
import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
+import org.apache.beam.sdk.util.TimerSpec;
import org.apache.beam.sdk.util.state.StateSpec;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TypeDescriptor;
@@ -70,6 +71,9 @@ public abstract class DoFnSignature {
@Nullable
public abstract LifecycleMethod teardown();
+ /** Timer declarations present on the {@link DoFn} class. Immutable. */
+ public abstract Map<String, TimerDeclaration> timerDeclarations();
+
/** Details about this {@link DoFn}'s {@link DoFn.GetInitialRestriction} method. */
@Nullable
public abstract GetInitialRestrictionMethod getInitialRestriction();
@@ -86,6 +90,10 @@ public abstract class DoFnSignature {
@Nullable
public abstract NewTrackerMethod newTracker();
+ /** Details about this {@link DoFn}'s {@link DoFn.OnTimer} methods. */
+ @Nullable
+ public abstract Map<String, OnTimerMethod> onTimerMethods();
+
static Builder builder() {
return new AutoValue_DoFnSignature.Builder();
}
@@ -104,6 +112,8 @@ public abstract class DoFnSignature {
abstract Builder setGetRestrictionCoder(GetRestrictionCoderMethod getRestrictionCoder);
abstract Builder setNewTracker(NewTrackerMethod newTracker);
abstract Builder setStateDeclarations(Map<String, StateDeclaration> stateDeclarations);
+ abstract Builder setTimerDeclarations(Map<String, TimerDeclaration> timerDeclarations);
+ abstract Builder setOnTimerMethods(Map<String, OnTimerMethod> onTimerMethods);
abstract DoFnSignature build();
}
@@ -160,6 +170,41 @@ public abstract class DoFnSignature {
}
}
+ /** Describes a {@link DoFn.OnTimer} method. */
+ @AutoValue
+ public abstract static class OnTimerMethod implements DoFnMethod {
+
+ /** The id on the method's {@link DoFn.TimerId} annotation. */
+ public abstract String id();
+
+ /** The annotated method itself. */
+ @Override
+ public abstract Method targetMethod();
+
+ /** Types of optional parameters of the annotated method, in the order they appear. */
+ public abstract List<Parameter> extraParameters();
+
+ static OnTimerMethod create(Method targetMethod, String id, List<Parameter> extraParameters) {
+ return new AutoValue_DoFnSignature_OnTimerMethod(
+ id, targetMethod, Collections.unmodifiableList(extraParameters));
+ }
+ }
+
+ /**
+ * Describes a timer declaration; a field of type {@link TimerSpec} annotated with
+ * {@DoFn.TimerId}.
+ */
+ @AutoValue
+ public abstract static class TimerDeclaration {
+ public abstract String id();
+ public abstract Field field();
+
+ static TimerDeclaration create(String id, Field field) {
+ return new AutoValue_DoFnSignature_TimerDeclaration(id, field);
+ }
+ }
+
+
/** Describes a {@link DoFn.StartBundle} or {@link DoFn.FinishBundle} method. */
@AutoValue
public abstract static class BundleMethod implements DoFnMethod {
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ccefc6f2/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java
index b87254e..9611dd4 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java
@@ -21,6 +21,7 @@ import static com.google.common.base.Preconditions.checkState;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
import com.google.common.reflect.TypeParameter;
import com.google.common.reflect.TypeToken;
import java.lang.annotation.Annotation;
@@ -33,6 +34,7 @@ import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
@@ -42,8 +44,12 @@ import javax.annotation.Nullable;
import javax.swing.plaf.nimbus.State;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignature.OnTimerMethod;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignature.StateDeclaration;
+import org.apache.beam.sdk.transforms.reflect.DoFnSignature.TimerDeclaration;
import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.util.TimerSpec;
import org.apache.beam.sdk.util.common.ReflectHelpers;
import org.apache.beam.sdk.util.state.StateSpec;
import org.apache.beam.sdk.values.PCollection;
@@ -96,6 +102,14 @@ public class DoFnSignatures {
}
errors.checkNotNull(inputT, "Unable to determine input type");
+ // Find the state and timer declarations in advance of validating
+ // method parameter lists
+ Map<String, StateDeclaration> stateDeclarations = analyzeStateDeclarations(errors, fnClass);
+ builder.setStateDeclarations(stateDeclarations);
+
+ Map<String, TimerDeclaration> timerDeclarations = analyzeTimerDeclarations(errors, fnClass);
+ builder.setTimerDeclarations(timerDeclarations);
+
Method processElementMethod =
findAnnotatedMethod(errors, DoFn.ProcessElement.class, fnClass, true);
Method startBundleMethod = findAnnotatedMethod(errors, DoFn.StartBundle.class, fnClass, false);
@@ -112,6 +126,41 @@ public class DoFnSignatures {
findAnnotatedMethod(errors, DoFn.GetRestrictionCoder.class, fnClass, false);
Method newTrackerMethod = findAnnotatedMethod(errors, DoFn.NewTracker.class, fnClass, false);
+ Collection<Method> onTimerMethods =
+ declaredMethodsWithAnnotation(DoFn.OnTimer.class, fnClass, DoFn.class);
+ HashMap<String, DoFnSignature.OnTimerMethod> onTimerMethodMap =
+ Maps.newHashMapWithExpectedSize(onTimerMethods.size());
+ for (Method onTimerMethod : onTimerMethods) {
+ String id = onTimerMethod.getAnnotation(DoFn.OnTimer.class).value();
+ errors.checkArgument(
+ timerDeclarations.containsKey(id),
+ "Callback %s is for for undeclared timer %s",
+ onTimerMethod,
+ id);
+
+ TimerDeclaration timerDecl = timerDeclarations.get(id);
+ errors.checkArgument(
+ timerDecl.field().getDeclaringClass().equals(onTimerMethod.getDeclaringClass()),
+ "Callback %s is for timer %s declared in a different class %s."
+ + " Timer callbacks must be declared in the same lexical scope as their timer",
+ onTimerMethod,
+ id,
+ timerDecl.field().getDeclaringClass().getCanonicalName());
+
+ onTimerMethodMap.put(id, OnTimerMethod.create(onTimerMethod, id, Collections.EMPTY_LIST));
+ }
+ builder.setOnTimerMethods(onTimerMethodMap);
+
+ // Check the converse - that all timers have a callback. This could be relaxed to only
+ // those timers used in methods, once method parameter lists support timers.
+ for (TimerDeclaration decl : timerDeclarations.values()) {
+ errors.checkArgument(
+ onTimerMethodMap.containsKey(decl.id()),
+ "No callback registered via %s for timer %s",
+ DoFn.OnTimer.class.getSimpleName(),
+ decl.id());
+ }
+
ErrorReporter processElementErrors =
errors.forMethod(DoFn.ProcessElement.class, processElementMethod);
DoFnSignature.ProcessElementMethod processElement =
@@ -183,8 +232,6 @@ public class DoFnSignatures {
builder.setIsBoundedPerElement(inferBoundedness(fnToken, processElement, errors));
- builder.setStateDeclarations(analyzeStateDeclarations(errors, fnClass));
-
DoFnSignature signature = builder.build();
// Additional validation for splittable DoFn's.
@@ -556,6 +603,50 @@ public class DoFnSignatures {
return DoFnSignature.SplitRestrictionMethod.create(m, restrictionT);
}
+ private static ImmutableMap<String, TimerDeclaration> analyzeTimerDeclarations(
+ ErrorReporter errors, Class<?> fnClazz) {
+ Map<String, DoFnSignature.TimerDeclaration> declarations = new HashMap<>();
+ for (Field field : declaredFieldsWithAnnotation(DoFn.TimerId.class, fnClazz, DoFn.class)) {
+ String id = field.getAnnotation(DoFn.TimerId.class).value();
+ validateTimerField(errors, declarations, id, field);
+ declarations.put(id, DoFnSignature.TimerDeclaration.create(id, field));
+ }
+
+ return ImmutableMap.copyOf(declarations);
+ }
+
+ /**
+ * Returns successfully if the field is valid, otherwise throws an exception via
+ * its {@link ErrorReporter} parameter describing validation failures for the
+ * timer declaration.
+ */
+ private static void validateTimerField(
+ ErrorReporter errors, Map<String, TimerDeclaration> declarations, String id, Field field) {
+ if (declarations.containsKey(id)) {
+ errors.throwIllegalArgument(
+ "Duplicate %s \"%s\", used on both of [%s] and [%s]",
+ DoFn.TimerId.class.getSimpleName(),
+ id,
+ field.toString(),
+ declarations.get(id).field().toString());
+ }
+
+ Class<?> timerSpecRawType = field.getType();
+ if (!(timerSpecRawType.equals(TimerSpec.class))) {
+ errors.throwIllegalArgument(
+ "%s annotation on non-%s field [%s]",
+ DoFn.TimerId.class.getSimpleName(),
+ TimerSpec.class.getSimpleName(),
+ field.toString());
+ }
+
+ if (!Modifier.isFinal(field.getModifiers())) {
+ errors.throwIllegalArgument(
+ "Non-final field %s annotated with %s. Timer declarations must be final.",
+ field.toString(), DoFn.TimerId.class.getSimpleName());
+ }
+ }
+
/** Generates a type token for {@code Coder<T>} given {@code T}. */
private static <T> TypeToken<Coder<T>> coderTypeOf(TypeToken<T> elementT) {
return new TypeToken<Coder<T>>() {}.where(new TypeParameter<T>() {}, elementT);
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ccefc6f2/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java
index 230e6ca..9813af5 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java
@@ -25,7 +25,11 @@ import com.google.common.reflect.TypeToken;
import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.coders.VarLongCoder;
import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.DoFn.TimerId;
import org.apache.beam.sdk.transforms.reflect.DoFnSignaturesTestUtils.FakeDoFn;
+import org.apache.beam.sdk.util.TimeDomain;
+import org.apache.beam.sdk.util.TimerSpec;
+import org.apache.beam.sdk.util.TimerSpecs;
import org.apache.beam.sdk.util.state.StateSpec;
import org.apache.beam.sdk.util.state.StateSpecs;
import org.apache.beam.sdk.util.state.ValueState;
@@ -132,6 +136,191 @@ public class DoFnSignaturesTest {
}
@Test
+ public void testTimerIdWithWrongType() throws Exception {
+ thrown.expect(IllegalArgumentException.class);
+ thrown.expectMessage("TimerId");
+ thrown.expectMessage("TimerSpec");
+ thrown.expectMessage("bizzle");
+ DoFnSignatures.INSTANCE.getSignature(
+ new DoFn<String, String>() {
+ @TimerId("foo")
+ private final String bizzle = "bazzle";
+
+ @ProcessElement
+ public void foo(ProcessContext context) {}
+ }.getClass());
+ }
+
+ @Test
+ public void testTimerIdNoCallback() throws Exception {
+ thrown.expect(IllegalArgumentException.class);
+ thrown.expectMessage("No callback registered");
+ thrown.expectMessage("my-timer-id");
+ DoFnSignature sig =
+ DoFnSignatures.INSTANCE.getSignature(
+ new DoFn<KV<String, Integer>, Long>() {
+ @TimerId("my-timer-id")
+ private final TimerSpec myfield1 = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ @ProcessElement
+ public void foo(ProcessContext context) {}
+ }.getClass());
+ }
+
+ @Test
+ public void testOnTimerNoDeclaration() throws Exception {
+ thrown.expect(IllegalArgumentException.class);
+ thrown.expectMessage("Callback");
+ thrown.expectMessage("undeclared timer");
+ thrown.expectMessage("onTimerFoo");
+ thrown.expectMessage("my-timer-id");
+ DoFnSignature sig =
+ DoFnSignatures.INSTANCE.getSignature(
+ new DoFn<KV<String, Integer>, Long>() {
+ @OnTimer("my-timer-id")
+ public void onTimerFoo() {}
+
+ @ProcessElement
+ public void foo(ProcessContext context) {}
+ }.getClass());
+ }
+
+ @Test
+ public void testOnTimerDeclaredInSuperclass() throws Exception {
+ thrown.expect(IllegalArgumentException.class);
+ thrown.expectMessage("Callback");
+ thrown.expectMessage("declared in a different class");
+ thrown.expectMessage("my-timer-id");
+ DoFnSignature sig =
+ DoFnSignatures.INSTANCE.getSignature(
+ new DoFnDeclaringMyTimerId() {
+ @OnTimer("my-timer-id")
+ public void onTimerFoo() {}
+
+ @ProcessElement
+ public void foo(ProcessContext context) {}
+ }.getClass());
+ }
+
+ @Test
+ public void testOnTimerDeclaredInSubclass() throws Exception {
+ thrown.expect(IllegalArgumentException.class);
+ thrown.expectMessage("Callback");
+ thrown.expectMessage("declared in a different class");
+ thrown.expectMessage("my-timer-id");
+ DoFnSignature sig =
+ DoFnSignatures.INSTANCE.getSignature(
+ new DoFnUsingMyTimerId() {
+ @TimerId("my-timer-id")
+ private final TimerSpec myfield1 = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ @ProcessElement
+ public void foo(ProcessContext context) {}
+ }.getClass());
+ }
+
+ /**
+ * In this particular test, the super class annotated both the timer and the callback,
+ * and the subclass overrides an abstract method. This is allowed.
+ */
+ @Test
+ public void testOnTimerDeclaredAndUsedInSuperclass() throws Exception {
+ DoFnSignature sig =
+ DoFnSignatures.INSTANCE.getSignature(
+ new DoFnOverridingAbstractCallback().getClass());
+
+ assertThat(sig.timerDeclarations().size(), equalTo(1));
+ assertThat(sig.onTimerMethods().size(), equalTo(1));
+
+ DoFnSignature.TimerDeclaration decl = sig.timerDeclarations().get("my-timer-id");
+ DoFnSignature.OnTimerMethod callback = sig.onTimerMethods().get("my-timer-id");
+
+ assertThat(
+ decl.field(),
+ equalTo(DoFnDeclaringMyTimerIdAndAbstractCallback.class.getDeclaredField("myTimerSpec")));
+
+ // The method we pull out is the superclass method; this is what allows validation to remain
+ // simple. The later invokeDynamic instruction causes it to invoke the actual implementation.
+ assertThat(
+ callback.targetMethod(),
+ equalTo(DoFnDeclaringMyTimerIdAndAbstractCallback.class.getDeclaredMethod("onMyTimer")));
+ }
+
+ @Test
+ public void testTimerIdDuplicate() throws Exception {
+ thrown.expect(IllegalArgumentException.class);
+ thrown.expectMessage("Duplicate");
+ thrown.expectMessage("TimerId");
+ thrown.expectMessage("my-timer-id");
+ thrown.expectMessage("myfield1");
+ thrown.expectMessage("myfield2");
+ DoFnSignature sig =
+ DoFnSignatures.INSTANCE.getSignature(
+ new DoFn<KV<String, Integer>, Long>() {
+ @TimerId("my-timer-id")
+ private final TimerSpec myfield1 = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ @TimerId("my-timer-id")
+ private final TimerSpec myfield2 = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ @ProcessElement
+ public void foo(ProcessContext context) {}
+ }.getClass());
+ }
+
+ @Test
+ public void testTimerIdNonFinal() throws Exception {
+ thrown.expect(IllegalArgumentException.class);
+ thrown.expectMessage("Timer declarations must be final");
+ thrown.expectMessage("Non-final field");
+ thrown.expectMessage("myfield");
+ DoFnSignature sig =
+ DoFnSignatures.INSTANCE.getSignature(
+ new DoFn<KV<String, Integer>, Long>() {
+ @TimerId("my-timer-id")
+ private TimerSpec myfield = TimerSpecs.timer(TimeDomain.PROCESSING_TIME);
+
+ @ProcessElement
+ public void foo(ProcessContext context) {}
+ }.getClass());
+ }
+
+ @Test
+ public void testSimpleTimerIdAnonymousDoFn() throws Exception {
+ DoFnSignature sig =
+ DoFnSignatures.INSTANCE.getSignature(
+ new DoFn<KV<String, Integer>, Long>() {
+ @TimerId("foo")
+ private final TimerSpec bizzle = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ @ProcessElement
+ public void foo(ProcessContext context) {}
+
+ @OnTimer("foo")
+ public void onFoo() {}
+ }.getClass());
+
+ assertThat(sig.timerDeclarations().size(), equalTo(1));
+ DoFnSignature.TimerDeclaration decl = sig.timerDeclarations().get("foo");
+
+ assertThat(decl.id(), equalTo("foo"));
+ assertThat(decl.field().getName(), equalTo("bizzle"));
+ }
+
+ @Test
+ public void testSimpleTimerIdNamedDoFn() throws Exception {
+ // Test classes at the bottom of the file
+ DoFnSignature sig =
+ DoFnSignatures.INSTANCE.signatureForDoFn(new DoFnForTestSimpleTimerIdNamedDoFn());
+
+ assertThat(sig.timerDeclarations().size(), equalTo(1));
+ DoFnSignature.TimerDeclaration decl = sig.timerDeclarations().get("foo");
+
+ assertThat(decl.id(), equalTo("foo"));
+ assertThat(
+ decl.field(), equalTo(DoFnForTestSimpleTimerIdNamedDoFn.class.getDeclaredField("bizzle")));
+ }
+
public void testStateIdWithWrongType() throws Exception {
thrown.expect(IllegalArgumentException.class);
thrown.expectMessage("StateId");
@@ -265,4 +454,53 @@ public class DoFnSignaturesTest {
@ProcessElement
public void foo(ProcessContext context) {}
}
+
+ private static class DoFnForTestSimpleTimerIdNamedDoFn extends DoFn<KV<String, Integer>, Long> {
+ @TimerId("foo")
+ private final TimerSpec bizzle = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ @ProcessElement
+ public void foo(ProcessContext context) {}
+
+ @OnTimer("foo")
+ public void onFoo() {}
+ }
+
+ private abstract static class DoFnDeclaringMyTimerId extends DoFn<KV<String, Integer>, Long> {
+ @TimerId("my-timer-id")
+ private final TimerSpec bizzle = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ @ProcessElement
+ public void foo(ProcessContext context) {}
+ }
+
+ private abstract static class DoFnUsingMyTimerId extends DoFn<KV<String, Integer>, Long> {
+ @OnTimer("my-timer-id")
+ public void onMyTimer() {}
+
+ @ProcessElement
+ public void foo(ProcessContext context) {}
+ }
+
+ private abstract static class DoFnDeclaringMyTimerIdAndAbstractCallback
+ extends DoFn<KV<String, Integer>, Long> {
+ @TimerId("my-timer-id")
+ private final TimerSpec myTimerSpec = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ @ProcessElement
+ public void foo(ProcessContext context) {}
+
+ @OnTimer("my-timer-id")
+ public abstract void onMyTimer();
+ }
+
+ private static class DoFnOverridingAbstractCallback extends
+ DoFnDeclaringMyTimerIdAndAbstractCallback {
+
+ @Override
+ public void onMyTimer() {}
+
+ @ProcessElement
+ public void foo(ProcessContext context) {}
+ }
}
[2/6] incubator-beam git commit: Fix condition on state/splittable
ParDo validation
Posted by ke...@apache.org.
Fix condition on state/splittable ParDo validation
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/08fdb304
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/08fdb304
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/08fdb304
Branch: refs/heads/master
Commit: 08fdb304a777d981fa3860fcd0270aa40f99da75
Parents: 507cba8
Author: Kenneth Knowles <kl...@google.com>
Authored: Fri Oct 14 14:19:59 2016 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Tue Oct 18 13:09:04 2016 -0700
----------------------------------------------------------------------
.../core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/08fdb304/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
index 4d34c88..776f768 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
@@ -564,7 +564,7 @@ public class ParDo {
}
// State is semantically incompatible with splitting
- if (!signature.stateDeclarations().isEmpty()) {
+ if (!signature.stateDeclarations().isEmpty() && signature.processElement().isSplittable()) {
throw new UnsupportedOperationException(
String.format("%s is splittable and uses state, but these are not compatible",
fn.getClass().getName()));
[5/6] incubator-beam git commit: Reject all timers in ParDo, for now
Posted by ke...@apache.org.
Reject all timers in ParDo, for now
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/f9712f2b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/f9712f2b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/f9712f2b
Branch: refs/heads/master
Commit: f9712f2bacb9aac9d5df5c6021bb3cfb59758806
Parents: ccefc6f
Author: Kenneth Knowles <kl...@google.com>
Authored: Tue Oct 18 13:09:57 2016 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Wed Oct 19 17:52:21 2016 -0700
----------------------------------------------------------------------
.../java/org/apache/beam/sdk/transforms/ParDo.java | 17 +++++++++++++++++
1 file changed, 17 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/f9712f2b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
index 776f768..8aa87e4 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
@@ -563,12 +563,29 @@ public class ParDo {
DoFn.class.getSimpleName()));
}
+ // To be removed when the features are complete and runners have their own adequate
+ // rejection logic
+ if (!signature.timerDeclarations().isEmpty()) {
+ throw new UnsupportedOperationException(
+ String.format("Found %s annotations on %s, but %s cannot yet be used with timers.",
+ DoFn.TimerId.class.getSimpleName(),
+ fn.getClass().getName(),
+ DoFn.class.getSimpleName()));
+ }
+
// State is semantically incompatible with splitting
if (!signature.stateDeclarations().isEmpty() && signature.processElement().isSplittable()) {
throw new UnsupportedOperationException(
String.format("%s is splittable and uses state, but these are not compatible",
fn.getClass().getName()));
}
+
+ // Timers are semantically incompatible with splitting
+ if (!signature.timerDeclarations().isEmpty() && signature.processElement().isSplittable()) {
+ throw new UnsupportedOperationException(
+ String.format("%s is splittable and uses timers, but these are not compatible",
+ fn.getClass().getName()));
+ }
}
private static <InputT, OutputT> OldDoFn<InputT, OutputT> adapt(DoFn<InputT, OutputT> fn) {
[4/6] incubator-beam git commit: Add Timer, TimerId,
TimerSpec and TimerSpecs
Posted by ke...@apache.org.
Add Timer, TimerId, TimerSpec and TimerSpecs
These are the simple specifications that describe
a timer without instantiating it. Consists only of
the time domain.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/1b780c28
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/1b780c28
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/1b780c28
Branch: refs/heads/master
Commit: 1b780c285f4be2f9780f3fdf39baa52bfbb387e2
Parents: 08fdb30
Author: Kenneth Knowles <kl...@google.com>
Authored: Thu Oct 13 13:41:38 2016 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Wed Oct 19 17:52:21 2016 -0700
----------------------------------------------------------------------
.../org/apache/beam/sdk/transforms/DoFn.java | 80 +++++++++++++++++++-
.../java/org/apache/beam/sdk/util/Timer.java | 56 ++++++++++++++
.../org/apache/beam/sdk/util/TimerSpec.java | 30 ++++++++
.../org/apache/beam/sdk/util/TimerSpecs.java | 41 ++++++++++
4 files changed, 204 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/1b780c28/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
index c86693b..8b3aaf8 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java
@@ -43,6 +43,8 @@ import org.apache.beam.sdk.transforms.reflect.DoFnInvoker;
import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.util.Timer;
+import org.apache.beam.sdk.util.TimerSpec;
import org.apache.beam.sdk.util.WindowingInternals;
import org.apache.beam.sdk.util.state.State;
import org.apache.beam.sdk.util.state.StateSpec;
@@ -399,12 +401,13 @@ public abstract class DoFn<InputT, OutputT> implements Serializable, HasDisplayD
/**
* Annotation for declaring and dereferencing state cells.
*
- * <p><i>Not currently supported by any runner</i>.
+ * <p><i>Not currently supported by any runner. When ready, the feature will work as described
+ * here.</i>
*
* <p>To declare a state cell, create a field of type {@link StateSpec} annotated with a {@link
* StateId}. To use the cell during processing, add a parameter of the appropriate {@link State}
- * subclass to your {@link ProcessElement @ProcessElement} method, and annotate it with {@link
- * StateId}. See the following code for an example:
+ * subclass to your {@link ProcessElement @ProcessElement} or {@link OnTimer @OnTimer} method, and
+ * annotate it with {@link StateId}. See the following code for an example:
*
* <pre>{@code
* new DoFn<KV<Key, Foo>, Baz>() {
@@ -440,6 +443,77 @@ public abstract class DoFn<InputT, OutputT> implements Serializable, HasDisplayD
}
/**
+ * Annotation for declaring and dereferencing timers.
+ *
+ * <p><i>Not currently supported by any runner. When ready, the feature will work as described
+ * here.</i>
+ *
+ * <p>To declare a timer, create a field of type {@link TimerSpec} annotated with a {@link
+ * TimerId}. To use the cell during processing, add a parameter of the type {@link Timer} to your
+ * {@link ProcessElement @ProcessElement} or {@link OnTimer @OnTimer} method, and annotate it with
+ * {@link TimerId}. See the following code for an example:
+ *
+ * <pre>{@code
+ * new DoFn<KV<Key, Foo>, Baz>() {
+ * @TimerId("my-timer-id")
+ * private final TimerSpec myTimer = TimerSpecs.timerForDomain(TimeDomain.EVENT_TIME);
+ *
+ * @ProcessElement
+ * public void processElement(
+ * ProcessContext c,
+ * @TimerId("my-timer-id") Timer myTimer) {
+ * myTimer.setForNowPlus(Duration.standardSeconds(...));
+ * }
+ *
+ * @OnTimer("my-timer-id")
+ * public void onMyTimer() {
+ * ...
+ * }
+ * }
+ * }</pre>
+ *
+ * <p>Timers are subject to the following validity conditions:
+ *
+ * <ul>
+ * <li>Each timer must have a distinct id.
+ * <li>Any timer referenced in a parameter must be declared.
+ * <li>Timer declarations must be final.
+ * <li>All declared timers must have a corresponding callback annotated with {@link
+ * OnTimer @OnTimer}.
+ * </ul>
+ */
+ @Documented
+ @Retention(RetentionPolicy.RUNTIME)
+ @Target({ElementType.FIELD, ElementType.PARAMETER})
+ @Experimental(Kind.TIMERS)
+ public @interface TimerId {
+ /** The timer ID. */
+ String value();
+ }
+
+ /**
+ * Annotation for registering a callback for a timer.
+ *
+ * <p><i>Not currently supported by any runner. When ready, the feature will work as described
+ * here.</i>
+ *
+ * <p>See the javadoc for {@link TimerId} for use in a full example.
+ *
+ * <p>The method annotated with {@code @OnTimer} may have parameters according to the same logic
+ * as {@link ProcessElement}, but limited to the {@link BoundedWindow}, {@link State} subclasses,
+ * and {@link Timer}. State and timer parameters must be annotated with their {@link StateId} and
+ * {@link TimerId} respectively.
+ */
+ @Documented
+ @Retention(RetentionPolicy.RUNTIME)
+ @Target(ElementType.METHOD)
+ @Experimental(Kind.TIMERS)
+ public @interface OnTimer {
+ /** The timer ID. */
+ String value();
+ }
+
+ /**
* Annotation for the method to use to prepare an instance for processing bundles of elements. The
* method annotated with this must satisfy the following constraints
* <ul>
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/1b780c28/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timer.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timer.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timer.java
new file mode 100644
index 0000000..556287d
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/Timer.java
@@ -0,0 +1,56 @@
+/*
+ * 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.util;
+
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.joda.time.Duration;
+
+/**
+ * A timer for a specified time domain that can be set to register the desire for further processing
+ * at particular time in its specified time domain.
+ *
+ * <p>See {@link TimeDomain} for details on the time domains available.
+ *
+ * <p>In a {@link DoFn}, a {@link Timer} is specified by a {@link TimerSpec} annotated with {@link
+ * DoFn.TimerId}.
+ *
+ * <p>An implementation of {@link Timer} is implicitly scoped - it may be scoped to a key and
+ * window, or a key, window, and trigger, etc.
+ *
+ * <p>A timer exists in one of two states: set or unset. A timer can be set only for a single time
+ * per scope.
+ *
+ * <p>Timer callbacks are not guaranteed to be called immediately according to the local view of the
+ * {@link TimeDomain}, but will be called at some time after the requested time, in timestamp
+ * order.
+ */
+@Experimental(Experimental.Kind.TIMERS)
+public interface Timer {
+ /**
+ * Sets or resets the time relative to the current time in the timer's {@link TimeDomain} at which
+ * this it should fire. If the timer was already set, resets it to the new requested time.
+ */
+ void setForNowPlus(Duration durationFromNow);
+
+ /**
+ * Unsets this timer. It is permitted to {@code cancel()} whether or not the timer was actually
+ * set.
+ */
+ void cancel();
+}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/1b780c28/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpec.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpec.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpec.java
new file mode 100644
index 0000000..5b7717b
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpec.java
@@ -0,0 +1,30 @@
+/*
+ * 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.util;
+
+import java.io.Serializable;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+
+/**
+ * A specification for a {@link Timer}. This includes its {@link TimeDomain}.
+ */
+@Experimental(Kind.TIMERS)
+public interface TimerSpec extends Serializable {
+ TimeDomain getTimeDomain();
+}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/1b780c28/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpecs.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpecs.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpecs.java
new file mode 100644
index 0000000..f289ccd
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/TimerSpecs.java
@@ -0,0 +1,41 @@
+/*
+ * 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.util;
+
+import com.google.auto.value.AutoValue;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+
+/**
+ * Static methods for working with {@link TimerSpec}.
+ */
+@Experimental(Kind.TIMERS)
+public class TimerSpecs {
+
+ public static TimerSpec timer(TimeDomain timeDomain) {
+ return new AutoValue_TimerSpecs_SimpleTimerSpec(timeDomain);
+ }
+
+ /**
+ * A straightforward POJO {@link TimerSpec}. Package-level access for AutoValue.
+ */
+ @AutoValue
+ abstract static class SimpleTimerSpec implements TimerSpec {
+ public abstract TimeDomain getTimeDomain();
+ }
+}
[6/6] incubator-beam git commit: This closes #1102
Posted by ke...@apache.org.
This closes #1102
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/30781932
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/30781932
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/30781932
Branch: refs/heads/master
Commit: 307819327a5fc2b21775b470ff566bade4a50f2f
Parents: 6c6f824 f9712f2
Author: Kenneth Knowles <kl...@google.com>
Authored: Wed Oct 19 17:53:43 2016 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Wed Oct 19 17:53:43 2016 -0700
----------------------------------------------------------------------
.../org/apache/beam/sdk/transforms/DoFn.java | 80 +++++-
.../org/apache/beam/sdk/transforms/ParDo.java | 19 +-
.../sdk/transforms/reflect/DoFnSignature.java | 45 ++++
.../sdk/transforms/reflect/DoFnSignatures.java | 95 +++++++-
.../java/org/apache/beam/sdk/util/Timer.java | 56 +++++
.../org/apache/beam/sdk/util/TimerSpec.java | 30 +++
.../org/apache/beam/sdk/util/TimerSpecs.java | 41 ++++
.../transforms/reflect/DoFnSignaturesTest.java | 244 ++++++++++++++++++-
8 files changed, 602 insertions(+), 8 deletions(-)
----------------------------------------------------------------------