protected abstract Source.Reader<T> createReader(@Nonnull FlinkSourceSplit<T> sourceSplit)
      throws IOException;
@Test
  public void testNumBytesInMetrics() throws Exception {
    final int numSplits = 2;
    final int numRecordsPerSplit = 10;
    List<FlinkSourceSplit<KV<Integer, Integer>>> splits =
        createSplits(numSplits, numRecordsPerSplit, 0);
    SourceTestMetrics.TestMetricGroup testMetricGroup = new SourceTestMetrics.TestMetricGroup();
    try (SourceReader<OutputT, FlinkSourceSplit<KV<Integer, Integer>>> reader =
        createReader(null, -1L, null, testMetricGroup)) {
      pollAndValidate(reader, splits, false);
    }
    assertEquals(numRecordsPerSplit * numSplits, testMetricGroup.numRecordsInCounter.getCount());
  }