SimpleParDoFn implements ParDoFn { @Override public void startBundle(Receiver... receivers) throws Exception { checkArgument( receivers.length == outputTupleTagsToReceiverIndices.size(), "unexpected number of receivers for DoFn"); this.receivers = receivers; if (hasStreamingSideInput) { reallyStartBundle(); } } SimpleParDoFn(
      PipelineOptions options,
      DoFnInstanceManager doFnInstanceManager,
      SideInputReader sideInputReader,
      TupleTag<OutputT> mainOutputTag,
      Map<TupleTag<?>, Integer> outputTupleTagsToReceiverIndices,
      DataflowExecutionContext.DataflowStepContext stepContext,
      DataflowOperationContext operationContext,
      DoFnSchemaInformation doFnSchemaInformation,
      Map<String, PCollectionView<?>> sideInputMapping,
      DoFnRunnerFactory runnerFactory); @Override void startBundle(Receiver... receivers); @Override @SuppressWarnings("unchecked") void processElement(Object untypedElem); @Override void processTimers(); @Override void finishBundle(); @Override void abort();  }
@Test @SuppressWarnings("AssertionFailureIgnored") public void testUnexpectedNumberOfReceivers() throws Exception { TestDoFn fn = new TestDoFn(Collections.emptyList()); DoFnInfo<?, ?> fnInfo = DoFnInfo.forFn( fn, WindowingStrategy.globalDefault(), null , null , MAIN_OUTPUT, DoFnSchemaInformation.create(), Collections.emptyMap()); TestReceiver receiver = new TestReceiver(); ParDoFn userParDoFn = new SimpleParDoFn<>( options, DoFnInstanceManagers.singleInstance(fnInfo), new EmptySideInputReader(), MAIN_OUTPUT, ImmutableMap.of(MAIN_OUTPUT, 0), BatchModeExecutionContext.forTesting(options, "testStage") .getStepContext(operationContext), operationContext, DoFnSchemaInformation.create(), Collections.emptyMap(), SimpleDoFnRunnerFactory.INSTANCE); try { userParDoFn.startBundle(); fail("should have failed"); } catch (Throwable exn) { assertThat(exn.toString(), containsString("unexpected number of receivers")); } try { userParDoFn.startBundle(receiver, receiver); fail("should have failed"); } catch (Throwable exn) { assertThat(exn.toString(), containsString("unexpected number of receivers")); } }