public static NativeReader<WindowedValue<?>> create(
      final CloudObject spec,
      final PipelineOptions options,
      DataflowExecutionContext executionContext)
      throws Exception {

    @SuppressWarnings("unchecked")
    final Source<Object> source = (Source<Object>) deserializeFromCloudSource(spec);

    if (source instanceof BoundedSource) {
      @SuppressWarnings({"unchecked", "rawtypes"})
      NativeReader<WindowedValue<?>> reader =
          (NativeReader)
              new NativeReader<WindowedValue<Object>>() {
                @Override
                public NativeReaderIterator<WindowedValue<Object>> iterator() throws IOException {
                  return new BoundedReaderIterator<>(
                      ((BoundedSource<Object>) source).createReader(options));
                }
              };
      return reader;
    } else if (source instanceof UnboundedSource) {
      @SuppressWarnings({"unchecked", "rawtypes"})
      NativeReader<WindowedValue<?>> reader =
          (NativeReader)
              new UnboundedReader<Object>(
                  options, spec, (StreamingModeExecutionContext) executionContext);
      return reader;
    } else {
      throw new IllegalArgumentException("Unexpected source kind: " + source.getClass());
    }
  }
@Test
  public void testFailedWorkItemsAbort() throws Exception {
    CounterSet counterSet = new CounterSet();
    StreamingModeExecutionStateRegistry executionStateRegistry =
        new StreamingModeExecutionStateRegistry();
    StreamingModeExecutionContext context =
        new StreamingModeExecutionContext(
            counterSet,
            COMPUTATION_ID,
            new ReaderCache(Duration.standardMinutes(1), Runnable::run),
            /*stateNameMap=*/ ImmutableMap.of(),
            WindmillStateCache.builder()
                .setSizeMb(options.getWorkerCacheMb())
                .build()
                .forComputation(COMPUTATION_ID),
            StreamingStepMetricsContainer.createRegistry(),
            new DataflowExecutionStateTracker(
                ExecutionStateSampler.newForTest(),
                executionStateRegistry.getState(
                    NameContext.forStage("stageName"), "other", null, NoopProfileScope.NOOP),
                counterSet,
                PipelineOptionsFactory.create(),
                "test-work-item-id"),
            executionStateRegistry,
            Long.MAX_VALUE);

    options.setNumWorkers(5);
    int maxElements = 100;
    DataflowPipelineDebugOptions debugOptions = options.as(DataflowPipelineDebugOptions.class);
    debugOptions.setUnboundedReaderMaxElements(maxElements);

    ByteString state = ByteString.EMPTY;
    Windmill.WorkItem workItem =
        Windmill.WorkItem.newBuilder()
            .setKey(ByteString.copyFromUtf8("0000000000000001")) // key is zero-padded index.
            .setWorkToken(0)
            .setCacheToken(1)
            .setSourceState(
                Windmill.SourceState.newBuilder().setState(state).build()) // Source state.
            .build();
    Work dummyWork =
        Work.create(
            workItem,
            Watermarks.builder().setInputDataWatermark(new Instant(0)).build(),
            Work.createProcessingContext(
                COMPUTATION_ID,
                new FakeGetDataClient(),
                ignored -> {},
                mock(HeartbeatSender.class)),
            Instant::now,
            Collections.emptyList());
    context.start(
        "key",
        dummyWork,
        mock(WindmillStateReader.class),
        mock(SideInputStateFetcher.class),
        OperationalLimits.builder().build(),
        Windmill.WorkItemCommitRequest.newBuilder());

    @SuppressWarnings({"unchecked", "rawtypes"})
    NativeReader<WindowedValue<ValueWithRecordId<KV<Integer, Integer>>>> reader =
        (NativeReader)
            WorkerCustomSources.create(
                (CloudObject)
                    serializeToCloudSource(new TestCountingSource(Integer.MAX_VALUE), options)
                        .getSpec(),
                options,
                context);

    NativeReaderIterator<WindowedValue<ValueWithRecordId<KV<Integer, Integer>>>> readerIterator =
        reader.iterator();
    int numReads = 0;
    while ((numReads == 0) ? readerIterator.start() : readerIterator.advance()) {
      WindowedValue<ValueWithRecordId<KV<Integer, Integer>>> value = readerIterator.getCurrent();
      assertEquals(KV.of(0, numReads), value.getValue().getValue());
      numReads++;
      // Fail the work item after reading two elements.
      if (numReads == 2) {
        dummyWork.setFailed();
      }
    }
    assertThat(numReads, equalTo(2));
  }