@Override
  protected Source.Reader<T> createReader(@Nonnull FlinkSourceSplit<T> sourceSplit)
      throws IOException {
    Source<T> beamSource = sourceSplit.getBeamSplitSource();
    return ((BoundedSource<T>) beamSource).createReader(pipelineOptions);
  }
@Test
  public void testIsAvailableOnSplitsAssignment() throws Exception {
    try (SourceReader<WindowedValue<KV<Integer, Integer>>, FlinkSourceSplit<KV<Integer, Integer>>>
        reader = createReader()) {
      reader.start();

      CompletableFuture<Void> future1 = reader.isAvailable();
      assertFalse("No split assigned yet, should not be available.", future1.isDone());

      // Data available on split assigned.
      reader.addSplits(createSplits(1, 1, 0));
      assertTrue("Adding a split should complete future1", future1.isDone());
      assertTrue("Data should be available with a live split.", reader.isAvailable().isDone());
    }
  }