@Override
  public void startBundle(Receiver... receivers) throws Exception {
    checkArgument(
        receivers.length == outputTupleTagsToReceiverIndices.size(),
        "unexpected number of receivers for DoFn");

    this.receivers = receivers;
    if (hasStreamingSideInput) {
      // There is non-trivial setup that needs to be performed for watermark propagation
      // even on empty bundles.
      reallyStartBundle();
    }
  }
@Test
  @SuppressWarnings("AssertionFailureIgnored")
  public void testUnexpectedNumberOfReceivers() throws Exception {
    TestDoFn fn = new TestDoFn(Collections.emptyList());
    DoFnInfo<?, ?> fnInfo =
        DoFnInfo.forFn(
            fn,
            WindowingStrategy.globalDefault(),
            null /* side input views */,
            null /* input coder */,
            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"));
    }
  }