@Override
  public NativeReader<?> create(
      CloudObject spec,
      @Nullable Coder<?> coder,
      @Nullable PipelineOptions options,
      @Nullable DataflowExecutionContext executionContext,
      DataflowOperationContext operationContext)
      throws Exception {
    @SuppressWarnings("unchecked")
    Coder<Object> typedCoder = (Coder<Object>) coder;
    return createTyped(spec, typedCoder, options, executionContext, operationContext);
  }
@Test
  public void testCreateConcatReaderWithOneSubSource() throws Exception {
    List<List<String>> allData = createInMemorySourceData(1, 10);

    Source source = createSourcesWithInMemorySources(allData);

    @SuppressWarnings("unchecked")
    NativeReader<String> reader =
        (NativeReader<String>) ReaderRegistry.defaultRegistry().create(source, null, null, null);
    assertNotNull(reader);

    List<String> expected = new ArrayList<>();
    for (List<String> data : allData) {
      expected.addAll(data);
    }
    assertThat(readAllFromReader(reader), containsInAnyOrder(expected.toArray()));
  }