DoFnSignatures { public static synchronized <FnT extends DoFn<?, ?>> DoFnSignature getSignature(Class<FnT> fn) { return signatureCache.computeIfAbsent(fn, k -> parseSignature(fn)); } private  DoFnSignatures(); static DoFnSignature signatureForDoFn(FnT fn); static synchronized DoFnSignature getSignature(Class<FnT> fn); static StateSpec<?> getStateSpecOrThrow(
      StateDeclaration stateDeclaration, DoFn<?, ?> target); static TimerSpec getTimerSpecOrThrow(
      TimerDeclaration timerDeclaration, DoFn<?, ?> target); static TimerSpec getTimerFamilySpecOrThrow(
      TimerFamilyDeclaration timerFamilyDeclaration, DoFn<?, ?> target); static boolean isSplittable(DoFn<?, ?> doFn); static boolean isStateful(DoFn<?, ?> doFn); static boolean usesMapState(DoFn<?, ?> doFn); static boolean usesSetState(DoFn<?, ?> doFn); static boolean usesOrderedListState(DoFn<?, ?> doFn); static boolean usesValueState(DoFn<?, ?> doFn); static boolean usesBagState(DoFn<?, ?> doFn); static boolean usesWatermarkHold(DoFn<?, ?> doFn); static boolean usesTimers(DoFn<?, ?> doFn); static boolean usesState(DoFn<?, ?> doFn); static boolean requiresTimeSortedInput(DoFn<?, ?> doFn);  }
@Test public void testBasicDoFnProcessContext() throws Exception { DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void process(ProcessContext c) {} }.getClass()); assertThat(sig.processElement().extraParameters().size(), equalTo(1)); assertThat( sig.processElement().extraParameters().get(0), instanceOf(ProcessContextParameter.class)); }
@Test public void testBasicDoFnAllParameters() throws Exception { DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void process( @Element String element, @Timestamp Instant timestamp, BoundedWindow window, PaneInfo paneInfo, OutputReceiver<String> receiver, PipelineOptions options, @SideInput("tag1") String input1, @SideInput("tag2") Integer input2, BundleFinalizer bundleFinalizer) {} }.getClass()); assertThat(sig.processElement().extraParameters().size(), equalTo(9)); assertThat(sig.processElement().extraParameters().get(0), instanceOf(ElementParameter.class)); assertThat(sig.processElement().extraParameters().get(1), instanceOf(TimestampParameter.class)); assertThat(sig.processElement().extraParameters().get(2), instanceOf(WindowParameter.class)); assertThat(sig.processElement().extraParameters().get(3), instanceOf(PaneInfoParameter.class)); assertThat( sig.processElement().extraParameters().get(4), instanceOf(OutputReceiverParameter.class)); assertThat( sig.processElement().extraParameters().get(5), instanceOf(PipelineOptionsParameter.class)); assertThat(sig.processElement().extraParameters().get(6), instanceOf(SideInputParameter.class)); assertThat(sig.processElement().extraParameters().get(7), instanceOf(SideInputParameter.class)); assertThat( sig.processElement().extraParameters().get(8), instanceOf(BundleFinalizerParameter.class)); }
@Test public void testBasicDoFnMultiOutputReceiver() throws Exception { DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void process(MultiOutputReceiver receiver) {} }.getClass()); assertThat(sig.processElement().extraParameters().size(), equalTo(1)); assertThat( sig.processElement().extraParameters().get(0), instanceOf(TaggedOutputReceiverParameter.class)); }
@Test public void testMismatchingElementType() throws Exception { DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void process(@Element Integer element) {} }.getClass()); assertThat(sig.processElement().extraParameters().size(), equalTo(1)); assertThat( sig.processElement().extraParameters().get(0), instanceOf(SchemaElementParameter.class)); }
@Test public void testWrongTimestampType() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("@Timestamp argument must have type org.joda.time.Instant"); DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void process(@Timestamp String timestamp) {} }.getClass()); }
@Test public void testWrongOutputReceiverType() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("OutputReceiver should be parameterized by java.lang.String"); DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void process(OutputReceiver<Integer> receiver) {} }.getClass()); }
@Test public void testRowParameterWithoutFieldAccess() { DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void process(@Element Row row) {} }.getClass()); assertFalse(sig.processElement().getSchemaElementParameters().isEmpty()); }
@Test public void testMultipleSchemaParameters() { DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void process( @Element Row row1, @Timestamp Instant ts, @Element Row row2, OutputReceiver<String> o, @Element Integer intParameter) {} }.getClass()); assertEquals(3, sig.processElement().getSchemaElementParameters().size()); assertEquals(0, sig.processElement().getSchemaElementParameters().get(0).index()); assertEquals( TypeDescriptors.rows(), sig.processElement().getSchemaElementParameters().get(0).elementT()); assertEquals(1, sig.processElement().getSchemaElementParameters().get(1).index()); assertEquals( TypeDescriptors.rows(), sig.processElement().getSchemaElementParameters().get(1).elementT()); assertEquals(2, sig.processElement().getSchemaElementParameters().get(2).index()); assertEquals( TypeDescriptors.integers(), sig.processElement().getSchemaElementParameters().get(2).elementT()); }
@Test public void testFieldAccess() throws IllegalAccessException { FieldAccessDescriptor descriptor = FieldAccessDescriptor.withFieldNames("foo", "bar"); DoFn<String, String> doFn = new DoFn<String, String>() { @FieldAccess("foo") final FieldAccessDescriptor fieldAccess = descriptor; @ProcessElement public void process(@FieldAccess("foo") @Element Row row) {} }; DoFnSignature sig = DoFnSignatures.getSignature(doFn.getClass()); assertThat(sig.fieldAccessDeclarations().get("foo"), notNullValue()); Field field = sig.fieldAccessDeclarations().get("foo").field(); assertThat(field.getName(), equalTo("fieldAccess")); assertThat(field.get(doFn), equalTo(descriptor)); assertFalse(sig.processElement().getSchemaElementParameters().isEmpty()); }
@Test public void testRowReceiver() { DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void process(OutputReceiver<Row> rowReceiver) {} }.getClass()); assertThat(sig.processElement().getMainOutputReceiver().isRowReceiver(), is(true)); }
@Test public void testRequiresStableInputProcessElement() throws Exception { DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement @RequiresStableInput public void process(ProcessContext c) {} }.getClass()); assertThat(sig.processElement().requiresStableInput(), is(true)); }
@Test public void testMultipleStartBundleElement() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("Found multiple methods annotated with @StartBundle"); thrown.expectMessage("bar()"); thrown.expectMessage("baz()"); thrown.expectMessage(getClass().getName() + "$"); DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void foo() {} @StartBundle public void bar() {} @StartBundle public void baz() {} }.getClass()); }
@Test public void testMultipleFinishBundleMethods() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("Found multiple methods annotated with @FinishBundle"); thrown.expectMessage("bar(FinishBundleContext)"); thrown.expectMessage("baz(FinishBundleContext)"); thrown.expectMessage(getClass().getName() + "$"); DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void foo(ProcessContext context) {} @FinishBundle public void bar(FinishBundleContext context) {} @FinishBundle public void baz(FinishBundleContext context) {} }.getClass()); }
@Test public void testPrivateStartBundle() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("startBundle()"); thrown.expectMessage("Must be public"); thrown.expectMessage(getClass().getName() + "$"); DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void processElement() {} @StartBundle void startBundle() {} }.getClass()); }
@Test public void testStartBundleWithAllParameters() throws Exception { DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void processElement() {} @StartBundle public void startBundle( StartBundleContext context, BundleFinalizer bundleFinalizer, PipelineOptions options) {} }.getClass()); assertThat(sig.startBundle().extraParameters().size(), equalTo(3)); assertThat( sig.startBundle().extraParameters().get(0), instanceOf(StartBundleContextParameter.class)); assertThat( sig.startBundle().extraParameters().get(1), instanceOf(BundleFinalizerParameter.class)); assertThat( sig.startBundle().extraParameters().get(2), instanceOf(PipelineOptionsParameter.class)); }
@Test public void testPrivateFinishBundle() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("finishBundle()"); thrown.expectMessage("Must be public"); thrown.expectMessage(getClass().getName() + "$"); DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void processElement() {} @FinishBundle void finishBundle() {} }.getClass()); }
@Test public void testFinishBundleWithAllParameters() throws Exception { DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void processElement() {} @FinishBundle public void finishBundle( FinishBundleContext context, BundleFinalizer bundleFinalizer, PipelineOptions pipelineOptions) {} }.getClass()); assertThat(sig.finishBundle().extraParameters().size(), equalTo(3)); assertThat( sig.finishBundle().extraParameters().get(0), instanceOf(FinishBundleContextParameter.class)); assertThat( sig.finishBundle().extraParameters().get(1), instanceOf(BundleFinalizerParameter.class)); assertThat( sig.finishBundle().extraParameters().get(2), instanceOf(PipelineOptionsParameter.class)); }
@Test public void testTimerIdWithWrongType() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("TimerId"); thrown.expectMessage("TimerSpec"); thrown.expectMessage("bizzle"); thrown.expectMessage(not(mentionsState())); DoFnSignatures.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-id"); thrown.expectMessage(not(mentionsState())); thrown.expectMessage(mentionsTimers()); DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @TimerId("my-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("onFoo"); thrown.expectMessage("my-id"); thrown.expectMessage(not(mentionsState())); thrown.expectMessage(mentionsTimers()); DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @OnTimer("my-id") public void onFoo() {} @ProcessElement public void foo(ProcessContext context) {} }.getClass()); }
@Test public void testOnTimerDeclaredInSuperclass() throws Exception { class DoFnDeclaringTimerAndProcessElement extends DoFn<KV<String, Integer>, Long> { public static final String TIMER_ID = "my-timer-id"; @TimerId(TIMER_ID) private final TimerSpec bizzle = TimerSpecs.timer(TimeDomain.EVENT_TIME); @ProcessElement public void foo(ProcessContext context) {} } DoFnDeclaringTimerAndProcessElement fn = new DoFnDeclaringTimerAndProcessElement() { @OnTimer(DoFnDeclaringTimerAndProcessElement.TIMER_ID) public void onTimerFoo() {} }; thrown.expect(IllegalArgumentException.class); thrown.expectMessage("Callback"); thrown.expectMessage("declared in a different class"); thrown.expectMessage(DoFnDeclaringTimerAndProcessElement.TIMER_ID); thrown.expectMessage(fn.getClass().getSimpleName()); thrown.expectMessage(not(mentionsState())); thrown.expectMessage(mentionsTimers()); DoFnSignatures.getSignature(fn.getClass()); }
@Test public void testUsageOfTimerDeclaredInSuperclass() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("process"); thrown.expectMessage("declared in a different class"); thrown.expectMessage(DoFnDeclaringTimerAndCallback.TIMER_ID); thrown.expectMessage(not(mentionsState())); thrown.expectMessage(mentionsTimers()); DoFnSignatures.getSignature( new DoFnDeclaringTimerAndCallback() { @ProcessElement public void process( ProcessContext context, @TimerId(DoFnDeclaringTimerAndCallback.TIMER_ID) Timer timer) {} }.getClass()); }
@Test public void testTimerParameterDuplicate() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("duplicate"); thrown.expectMessage("my-id"); thrown.expectMessage("myProcessElement"); thrown.expectMessage("index 2"); thrown.expectMessage(not(mentionsState())); DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @TimerId("my-id") private final TimerSpec myfield = TimerSpecs.timer(TimeDomain.PROCESSING_TIME); @ProcessElement public void myProcessElement( ProcessContext context, @TimerId("my-id") Timer one, @TimerId("my-id") Timer two) {} @OnTimer("my-id") public void onWhatever() {} }.getClass()); }
@Test public void testOnTimerDeclaredInSubclass() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("Callback"); thrown.expectMessage("declared in a different class"); thrown.expectMessage(DoFnWithOnlyCallback.TIMER_ID); thrown.expectMessage(not(mentionsState())); thrown.expectMessage(mentionsTimers()); DoFnSignatures.getSignature( new DoFnWithOnlyCallback() { @TimerId(DoFnWithOnlyCallback.TIMER_ID) private final TimerSpec myfield1 = TimerSpecs.timer(TimeDomain.EVENT_TIME); @ProcessElement @Override public void foo(ProcessContext context) {} }.getClass()); }
@Test public void testWindowParamOnTimer() throws Exception { final String timerId = "some-timer-id"; final String timerDeclarationId = TimerDeclaration.PREFIX + timerId; DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @TimerId(timerId) private final TimerSpec myfield1 = TimerSpecs.timer(TimeDomain.EVENT_TIME); @ProcessElement public void process(ProcessContext c) {} @OnTimer(timerId) public void onTimer(BoundedWindow w) {} }.getClass()); assertThat(sig.onTimerMethods().get(timerDeclarationId).extraParameters().size(), equalTo(1)); assertThat( sig.onTimerMethods().get(timerDeclarationId).extraParameters().get(0), instanceOf(WindowParameter.class)); }
@Test public void testAllParamsOnTimer() throws Exception { final String timerId = "some-timer-id"; final String timerDeclarationId = TimerDeclaration.PREFIX + timerId; DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @TimerId(timerId) private final TimerSpec myfield1 = TimerSpecs.timer(TimeDomain.EVENT_TIME); @ProcessElement public void process(ProcessContext c) {} @OnTimer(timerId) public void onTimer( @Timestamp Instant timestamp, TimeDomain timeDomain, BoundedWindow w) {} }.getClass()); assertThat(sig.onTimerMethods().get(timerDeclarationId).extraParameters().size(), equalTo(3)); assertThat( sig.onTimerMethods().get(timerDeclarationId).extraParameters().get(0), instanceOf(TimestampParameter.class)); assertThat( sig.onTimerMethods().get(timerDeclarationId).extraParameters().get(1), instanceOf(TimeDomainParameter.class)); assertThat( sig.onTimerMethods().get(timerDeclarationId).extraParameters().get(2), instanceOf(WindowParameter.class)); }
@Test public void testPipelineOptionsParameter() throws Exception { DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void process(ProcessContext c, PipelineOptions options) {} }.getClass()); assertThat( sig.processElement().extraParameters(), Matchers.hasItem(instanceOf(Parameter.PipelineOptionsParameter.class))); }
@Test public void testDeclAndUsageOfTimerInSuperclass() throws Exception { DoFnSignature sig = DoFnSignatures.getSignature(new DoFnOverridingAbstractTimerUse().getClass()); assertThat(sig.timerDeclarations().size(), equalTo(1)); assertThat(sig.processElement().extraParameters().size(), equalTo(2)); DoFnSignature.TimerDeclaration decl = sig.timerDeclarations() .get(TimerDeclaration.PREFIX + DoFnOverridingAbstractTimerUse.TIMER_ID); TimerParameter timerParam = (TimerParameter) sig.processElement().extraParameters().get(1); assertThat( decl.field(), equalTo(DoFnDeclaringTimerAndAbstractUse.class.getDeclaredField("myTimerSpec"))); assertThat(timerParam.referent(), equalTo(decl)); }
@Test public void testOnTimerDeclaredAndUsedInSuperclass() throws Exception { DoFnSignature sig = DoFnSignatures.getSignature(new DoFnOverridingAbstractCallback().getClass()); assertThat(sig.timerDeclarations().size(), equalTo(1)); assertThat(sig.onTimerMethods().size(), equalTo(1)); DoFnSignature.TimerDeclaration decl = sig.timerDeclarations() .get(TimerDeclaration.PREFIX + DoFnDeclaringTimerAndAbstractCallback.TIMER_ID); DoFnSignature.OnTimerMethod callback = sig.onTimerMethods() .get(TimerDeclaration.PREFIX + DoFnDeclaringTimerAndAbstractCallback.TIMER_ID); assertThat( decl.field(), equalTo(DoFnDeclaringTimerAndAbstractCallback.class.getDeclaredField("myTimerSpec"))); assertThat( callback.targetMethod(), equalTo(DoFnDeclaringTimerAndAbstractCallback.class.getDeclaredMethod("onMyTimer"))); }
@Test public void testTimerIdDuplicate() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("Duplicate"); thrown.expectMessage("TimerId"); thrown.expectMessage("my-id"); thrown.expectMessage("myfield1"); thrown.expectMessage("myfield2"); thrown.expectMessage(not(containsString("State"))); thrown.expectMessage(mentionsTimers()); DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @TimerId("my-id") private final TimerSpec myfield1 = TimerSpecs.timer(TimeDomain.EVENT_TIME); @TimerId("my-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"); thrown.expectMessage(not(containsString("State"))); thrown.expectMessage(mentionsTimers()); DoFnSignatures.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.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()); final String timerDeclarationId = TimerDeclaration.PREFIX + "foo"; assertThat(sig.timerDeclarations().size(), equalTo(1)); DoFnSignature.TimerDeclaration decl = sig.timerDeclarations().get(timerDeclarationId); assertThat(decl.id(), equalTo(timerDeclarationId)); assertThat(decl.field().getName(), equalTo("bizzle")); }
@Test public void testSimpleTimerWithContext() throws Exception { DoFnSignature sig = DoFnSignatures.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(OnTimerContext c) {} }.getClass()); final String timerDeclarationId = TimerDeclaration.PREFIX + "foo"; assertThat(sig.timerDeclarations().size(), equalTo(1)); DoFnSignature.TimerDeclaration decl = sig.timerDeclarations().get(timerDeclarationId); assertThat(decl.id(), equalTo(timerDeclarationId)); assertThat(decl.field().getName(), equalTo("bizzle")); assertThat( sig.onTimerMethods().get(timerDeclarationId).extraParameters().get(0), equalTo((Parameter) Parameter.onTimerContext())); }
@Test public void testProcessElementWithOnTimerContextRejected() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("@" + DoFn.ProcessElement.class.getSimpleName()); thrown.expectMessage(DoFn.OnTimerContext.class.getSimpleName()); DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @TimerId("foo") private final TimerSpec bizzle = TimerSpecs.timer(TimeDomain.EVENT_TIME); @ProcessElement public void foo(ProcessContext context, OnTimerContext bogus) {} @OnTimer("foo") public void onFoo() {} }.getClass()); }
@Test public void testStateIdWithWrongType() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("StateId"); thrown.expectMessage("StateSpec"); thrown.expectMessage(not(mentionsTimers())); DoFnSignatures.getSignature( new DoFn<String, String>() { @StateId("foo") private final String bizzle = "bazzle"; @ProcessElement public void foo(ProcessContext context) {} }.getClass()); }
@Test public void testStateIdDuplicate() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("Duplicate"); thrown.expectMessage("StateId"); thrown.expectMessage("my-id"); thrown.expectMessage("myfield1"); thrown.expectMessage("myfield2"); thrown.expectMessage(not(mentionsTimers())); DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @StateId("my-id") private final StateSpec<ValueState<Integer>> myfield1 = StateSpecs.value(VarIntCoder.of()); @StateId("my-id") private final StateSpec<ValueState<Long>> myfield2 = StateSpecs.value(VarLongCoder.of()); @ProcessElement public void foo(ProcessContext context) {} }.getClass()); }
@Test public void testStateIdNonFinal() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("State declarations must be final"); thrown.expectMessage("Non-final field"); thrown.expectMessage("myfield"); thrown.expectMessage(not(mentionsTimers())); DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @StateId("my-id") private StateSpec<ValueState<Integer>> myfield = StateSpecs.value(VarIntCoder.of()); @ProcessElement public void foo(ProcessContext context) {} }.getClass()); }
@Test public void testStateParameterNoAnnotation() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("missing StateId annotation"); thrown.expectMessage("myProcessElement"); thrown.expectMessage("index 1"); thrown.expectMessage(not(mentionsTimers())); DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @ProcessElement public void myProcessElement(ProcessContext context, ValueState<Integer> noAnnotation) {} }.getClass()); }
@Test public void testStateParameterUndeclared() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("undeclared"); thrown.expectMessage("my-id"); thrown.expectMessage("myProcessElement"); thrown.expectMessage("index 1"); thrown.expectMessage(not(mentionsTimers())); DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @ProcessElement public void myProcessElement( ProcessContext context, @StateId("my-id") ValueState<Integer> undeclared) {} }.getClass()); }
@Test public void testStateParameterAlwaysFetched() { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("ReadableStates"); DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @StateId("my-id") private final StateSpec<MapState<Integer, Integer>> myfield = StateSpecs.map(VarIntCoder.of(), VarIntCoder.of()); @ProcessElement public void myProcessElement( ProcessContext context, @AlwaysFetched @StateId("my-id") MapState<Integer, Integer> one) {} }.getClass()); StateParameter stateParameter = (StateParameter) sig.processElement().extraParameters().get(1); assertTrue(stateParameter.alwaysFetched()); }
@Test public void testStateParameterAlwaysFetchNonReadableState() { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("ReadableStates"); DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @StateId("my-id") private final StateSpec<MapState<Integer, Integer>> myfield = StateSpecs.map(VarIntCoder.of(), VarIntCoder.of()); @ProcessElement public void myProcessElement( ProcessContext context, @AlwaysFetched @StateId("my-id") MapState<Integer, Integer> one) {} }.getClass()); }
@Test public void testStateParameterDuplicate() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("duplicate"); thrown.expectMessage("my-id"); thrown.expectMessage("myProcessElement"); thrown.expectMessage("index 2"); thrown.expectMessage(not(mentionsTimers())); DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @StateId("my-id") private final StateSpec<ValueState<Integer>> myfield = StateSpecs.value(VarIntCoder.of()); @ProcessElement public void myProcessElement( ProcessContext context, @StateId("my-id") ValueState<Integer> one, @StateId("my-id") ValueState<Integer> two) {} }.getClass()); }
@Test public void testStateParameterWrongStateType() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("WatermarkHoldState"); thrown.expectMessage("reference to"); thrown.expectMessage("supertype"); thrown.expectMessage("ValueState"); thrown.expectMessage("my-id"); thrown.expectMessage("myProcessElement"); thrown.expectMessage("index 1"); thrown.expectMessage(not(mentionsTimers())); DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @StateId("my-id") private final StateSpec<ValueState<Integer>> myfield = StateSpecs.value(VarIntCoder.of()); @ProcessElement public void myProcessElement( ProcessContext context, @StateId("my-id") WatermarkHoldState watermark) {} }.getClass()); }
@Test public void testStateParameterWrongGenericType() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("ValueState<String>"); thrown.expectMessage("reference to"); thrown.expectMessage("supertype"); thrown.expectMessage("ValueState<Integer>"); thrown.expectMessage("my-id"); thrown.expectMessage("myProcessElement"); thrown.expectMessage("index 1"); thrown.expectMessage(not(mentionsTimers())); DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @StateId("my-id") private final StateSpec<ValueState<Integer>> myfield = StateSpecs.value(VarIntCoder.of()); @ProcessElement public void myProcessElement( ProcessContext context, @StateId("my-id") ValueState<String> stringState) {} }.getClass()); }
@Test public void testGoodStateParameterSuperclassStateType() throws Exception { DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @StateId("my-id") private final StateSpec<CombiningState<Integer, int[], Integer>> state = StateSpecs.combining(Sum.ofIntegers()); @ProcessElement public void myProcessElement( ProcessContext context, @StateId("my-id") GroupingState<Integer, Integer> groupingState) {} }.getClass()); }
@Test public void testSimpleStateIdAnonymousDoFn() throws Exception { DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<KV<String, Integer>, Long>() { @StateId("foo") private final StateSpec<ValueState<Integer>> bizzle = StateSpecs.value(VarIntCoder.of()); @ProcessElement public void foo(ProcessContext context) {} }.getClass()); assertThat(sig.stateDeclarations().size(), equalTo(1)); DoFnSignature.StateDeclaration decl = sig.stateDeclarations().get("foo"); assertThat(decl.id(), equalTo("foo")); assertThat(decl.field().getName(), equalTo("bizzle")); assertThat( decl.stateType(), Matchers.<TypeDescriptor<?>>equalTo(new TypeDescriptor<ValueState<Integer>>() {})); }
@Test public void testUsageOfStateDeclaredInSuperclass() throws Exception { DoFnDeclaringState fn = new DoFnDeclaringState() { @ProcessElement public void process( ProcessContext context, @StateId(DoFnDeclaringState.STATE_ID) ValueState<Integer> state) {} }; thrown.expect(IllegalArgumentException.class); thrown.expectMessage("process"); thrown.expectMessage("declared in a different class"); thrown.expectMessage(DoFnDeclaringState.STATE_ID); thrown.expectMessage(fn.getClass().getSimpleName()); DoFnSignatures.getSignature(fn.getClass()); }
@Test public void testDeclOfStateUsedInSuperclass() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("process"); thrown.expectMessage("declared in a different class"); thrown.expectMessage(DoFnUsingState.STATE_ID); DoFnSignatures.getSignature( new DoFnUsingState() { @StateId(DoFnUsingState.STATE_ID) private final StateSpec<ValueState<Integer>> spec = StateSpecs.value(VarIntCoder.of()); }.getClass()); }
@Test public void testDeclAndUsageOfStateInSuperclass() throws Exception { class DoFnOverridingAbstractStateUse extends DoFnDeclaringStateAndAbstractUse { @Override public void processWithState(ProcessContext c, ValueState<String> state) {} } DoFnSignature sig = DoFnSignatures.getSignature(new DoFnOverridingAbstractStateUse().getClass()); assertThat(sig.stateDeclarations().size(), equalTo(1)); assertThat(sig.processElement().extraParameters().size(), equalTo(2)); DoFnSignature.StateDeclaration decl = sig.stateDeclarations().get(DoFnOverridingAbstractStateUse.STATE_ID); StateParameter stateParam = (StateParameter) sig.processElement().extraParameters().get(1); assertFalse(stateParam.alwaysFetched()); assertThat( decl.field(), equalTo(DoFnDeclaringStateAndAbstractUse.class.getDeclaredField("myStateSpec"))); assertThat(stateParam.referent(), equalTo(decl)); }
@Test public void testOnWindowExpirationMultipleAnnotation() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("Found multiple methods annotated with @OnWindowExpiration"); thrown.expectMessage("bar()"); thrown.expectMessage("baz()"); thrown.expectMessage(getClass().getName() + "$"); DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void foo() {} @OnWindowExpiration public void bar() {} @OnWindowExpiration public void baz() {} }.getClass()); }
@Test public void testOnWindowExpirationMustBePublic() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("OnWindowExpiration"); thrown.expectMessage("Must be public"); thrown.expectMessage("bar()"); DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void foo() {} @OnWindowExpiration void bar() {} }.getClass()); }
@Test public void testOnWindowExpirationDisallowedParameter() throws Exception { thrown.expect(IllegalArgumentException.class); thrown.expectMessage("Illegal parameter type"); thrown.expectMessage("TimerParameter"); thrown.expectMessage("myTimer"); DoFnSignatures.getSignature( new DoFn<String, String>() { @TimerId("foo") private final TimerSpec myTimer = TimerSpecs.timer(TimeDomain.EVENT_TIME); @ProcessElement public void foo() {} @OnTimer("foo") public void onFoo() {} @OnWindowExpiration public void bar(@TimerId("foo") Timer t) {} }.getClass()); }
@Test public void testOnWindowExpirationNoParam() { DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @ProcessElement public void process(ProcessContext c) {} @OnWindowExpiration public void bar() {} }.getClass()); assertThat(sig.onWindowExpiration().extraParameters().size(), equalTo(0)); }
@Test public void testOnWindowExpirationWithAllowedParams() { DoFnSignature sig = DoFnSignatures.getSignature( new DoFn<String, String>() { @StateId("foo") private final StateSpec<ValueState<Integer>> bizzle = StateSpecs.value(VarIntCoder.of()); @ProcessElement public void process(ProcessContext c) {} @OnWindowExpiration public void bar( BoundedWindow b, @StateId("foo") ValueState<Integer> s, PipelineOptions p, OutputReceiver<String> o, MultiOutputReceiver m) {} }.getClass()); List<Parameter> params = sig.onWindowExpiration().extraParameters(); assertThat(params.size(), equalTo(5)); assertThat(params.get(0), instanceOf(WindowParameter.class)); assertThat(params.get(1), instanceOf(StateParameter.class)); assertThat(params.get(2), instanceOf(PipelineOptionsParameter.class)); assertThat(params.get(3), instanceOf(OutputReceiverParameter.class)); assertThat(params.get(4), instanceOf(TaggedOutputReceiverParameter.class)); }