@Override
  public void start() {
    // We request a split only if we did not get splits during the checkpoint restore.
    // Otherwise, reader restarts will keep requesting more and more splits.
    if (getNumberOfCurrentlyAssignedSplits() == 0) {
      requestSplit(Collections.emptyList());
    }
  }
@Test
  public void testReaderMetrics() throws Exception {
    TestingReaderOutput<RowData> readerOutput = new TestingReaderOutput<>();
    TestingMetricGroup metricGroup = new TestingMetricGroup();
    TestingReaderContext readerContext = new TestingReaderContext(new Configuration(), metricGroup);
    IcebergSourceReader reader = createReader(metricGroup, readerContext, null);
    reader.start();

    testOneSplitFetcher(reader, readerOutput, metricGroup, 1);
    testOneSplitFetcher(reader, readerOutput, metricGroup, 2);
  }