SimpleDoFnRunner implements DoFnRunner<InputT, OutputT> { @Override public void startBundle() { try { invoker.invokeStartBundle(new DoFnStartBundleArgumentProvider()); } catch (Throwable t) { throw wrapUserCodeException(t); } } SimpleDoFnRunner(
      PipelineOptions options,
      DoFn<InputT, OutputT> fn,
      SideInputReader sideInputReader,
      OutputManager outputManager,
      TupleTag<OutputT> mainOutputTag,
      List<TupleTag<?>> additionalOutputTags,
      StepContext stepContext,
      @Nullable Coder<InputT> inputCoder,
      Map<TupleTag<?>, Coder<?>> outputCoders,
      WindowingStrategy<?, ?> windowingStrategy,
      DoFnSchemaInformation doFnSchemaInformation,
      Map<String, PCollectionView<?>> sideInputMapping); @Override DoFn<InputT, OutputT> getFn(); @Override void startBundle(); @Override void processElement(WindowedValue<InputT> compressedElem); @Override void onTimer(
      String timerId,
      String timerFamilyId,
      KeyT key,
      BoundedWindow window,
      Instant timestamp,
      Instant outputTimestamp,
      TimeDomain timeDomain); @Override void finishBundle(); @Override void onWindowExpiration(BoundedWindow window, Instant timestamp, KeyT key);  }
@Test public void testStartBundleExceptionsWrappedAsUserCodeException() { ThrowingDoFn fn = new ThrowingDoFn(); DoFnRunner<String, String> runner = new SimpleDoFnRunner<>( null, fn, NullSideInputReader.empty(), null, null, Collections.emptyList(), mockStepContext, null, Collections.emptyMap(), WindowingStrategy.of(new GlobalWindows()), DoFnSchemaInformation.create(), Collections.emptyMap()); thrown.expect(UserCodeException.class); thrown.expectCause(is(fn.exceptionToThrow)); runner.startBundle(); }