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 testReadUnboundedReader() throws Exception {
    CounterSet counterSet = new CounterSet();
    StreamingModeExecutionStateRegistry executionStateRegistry =
        new StreamingModeExecutionStateRegistry();
    ReaderCache readerCache = new ReaderCache(Duration.standardMinutes(1), Runnable::run);
    StreamingModeExecutionContext context =
        new StreamingModeExecutionContext(
            counterSet,
            COMPUTATION_ID,
            readerCache,
            /*stateNameMap=*/ ImmutableMap.of(),
            /*stateCache=*/ null,
            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 = 10;
    DataflowPipelineDebugOptions debugOptions = options.as(DataflowPipelineDebugOptions.class);
    debugOptions.setUnboundedReaderMaxElements(maxElements);

    ByteString state = ByteString.EMPTY;
    for (int i = 0; i < 10 * maxElements;
    /* Incremented in inner loop */ ) {
      // Initialize streaming context with state from previous iteration.
      context.start(
          "key",
          createMockWork(
              Windmill.WorkItem.newBuilder()
                  .setKey(ByteString.copyFromUtf8("0000000000000001")) // key is zero-padded index.
                  .setWorkToken(i) // Must be increasing across activations for cache to be used.
                  .setCacheToken(1)
                  .setSourceState(
                      Windmill.SourceState.newBuilder().setState(state).build()) // Source state.
                  .build(),
              Watermarks.builder().setInputDataWatermark(new Instant(0)).build()),
          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);

      // Verify data.
      Instant beforeReading = Instant.now();
      int numReadOnThisIteration = 0;
      for (WindowedValue<ValueWithRecordId<KV<Integer, Integer>>> value :
          ReaderUtils.readAllFromReader(reader)) {
        assertEquals(KV.of(0, i), value.getValue().getValue());
        assertArrayEquals(
            encodeToByteArray(KvCoder.of(VarIntCoder.of(), VarIntCoder.of()), KV.of(0, i)),
            value.getValue().getId());
        assertThat(value.getWindows(), contains((BoundedWindow) GlobalWindow.INSTANCE));
        assertEquals(i, value.getTimestamp().getMillis());
        i++;
        numReadOnThisIteration++;
      }
      Instant afterReading = Instant.now();
      long maxReadSec = debugOptions.getUnboundedReaderMaxReadTimeSec();
      assertThat(
          new Duration(beforeReading, afterReading).getStandardSeconds(),
          lessThanOrEqualTo(maxReadSec + 1));
      assertThat(
          numReadOnThisIteration, lessThanOrEqualTo(debugOptions.getUnboundedReaderMaxElements()));

      // Extract and verify state modifications.
      context.flushState();
      state = context.getOutputBuilder().getSourceStateUpdates().getState();
      // CountingSource's watermark is the last record + 1.  i is now one past the last record,
      // so the expected watermark is i millis.
      assertEquals(
          TimeUnit.MILLISECONDS.toMicros(i), context.getOutputBuilder().getSourceWatermark());
      assertEquals(
          1, context.getOutputBuilder().getSourceStateUpdates().getFinalizeIdsList().size());

      assertNotNull(
          readerCache.acquireReader(
              context.getComputationKey(),
              context.getWorkItem().getCacheToken(),
              context.getWorkToken() + 1));
      assertEquals(7L, context.getBacklogBytes());
    }
  }