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
  @SuppressWarnings("unchecked")
  public void testProgressAndSourceSplitTranslation() throws Exception {
    // Same as previous test, but now using BasicSerializableSourceFormat wrappers.
    // We know that the underlying reader behaves correctly (because of the previous test),
    // now check that we are wrapping it correctly.
    NativeReader<WindowedValue<Integer>> reader =
        (NativeReader<WindowedValue<Integer>>)
            ReaderRegistry.defaultRegistry()
                .create(
                    translateIOToCloudSource(CountingSource.upTo(10), options),
                    options,
                    null, // executionContext
                    TestOperationContext.create());
    try (NativeReader.NativeReaderIterator<WindowedValue<Integer>> iterator = reader.iterator()) {
      assertTrue(iterator.start());
      assertEquals(valueInGlobalWindow(0L), iterator.getCurrent());
      assertEquals(
          0.0,
          readerProgressToCloudProgress(iterator.getProgress()).getFractionConsumed().doubleValue(),
          1e-6);
      assertTrue(iterator.advance());
      assertEquals(valueInGlobalWindow(1L), iterator.getCurrent());
      assertEquals(
          0.1,
          readerProgressToCloudProgress(iterator.getProgress()).getFractionConsumed().doubleValue(),
          1e-6);
      assertTrue(iterator.advance());
      assertEquals(valueInGlobalWindow(2L), iterator.getCurrent());

      assertNull(iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtFraction(0)));
      assertNull(iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtFraction(0.1f)));
      WorkerCustomSources.BoundedSourceSplit<Integer> sourceSplit =
          (WorkerCustomSources.BoundedSourceSplit<Integer>)
              iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtFraction(0.5f));
      assertNotNull(sourceSplit);
      assertThat(readFromSource(sourceSplit.primary, options), contains(0L, 1L, 2L, 3L, 4L));
      assertThat(readFromSource(sourceSplit.residual, options), contains(5L, 6L, 7L, 8L, 9L));

      sourceSplit =
          (WorkerCustomSources.BoundedSourceSplit<Integer>)
              iterator.requestDynamicSplit(ReaderTestUtils.splitRequestAtFraction(0.8f));
      assertNotNull(sourceSplit);
      assertThat(readFromSource(sourceSplit.primary, options), contains(0L, 1L, 2L, 3L));
      assertThat(readFromSource(sourceSplit.residual, options), contains(4L));

      assertTrue(iterator.advance());
      assertEquals(valueInGlobalWindow(3L), iterator.getCurrent());
      assertFalse(iterator.advance());
    }
  }