@Override
  public void addSplitsBack(List<FlinkSourceSplit<T>> splits, int subtaskId) {
    LOG.info("Adding splits {} back from subtask {}", splits, subtaskId);
    List<FlinkSourceSplit<T>> splitsForSubtask =
        pendingSplits.computeIfAbsent(subtaskId, ignored -> new ArrayList<>());
    splitsForSubtask.addAll(splits);
  }
@Test
  public void testAddSplitsBack() throws IOException {
    final int numSubtasks = 2;
    final int numSplits = 10;
    final int totalNumRecords = 10;
    TestingSplitEnumeratorContext<FlinkSourceSplit<KV<Integer, Integer>>> testContext =
        new TestingSplitEnumeratorContext<>(numSubtasks);
    TestBoundedCountingSource testSource =
        new TestBoundedCountingSource(numSplits, totalNumRecords);
    try (FlinkSourceSplitEnumerator<KV<Integer, Integer>> splitEnumerator =
        new FlinkSourceSplitEnumerator<>(
            testContext, testSource, FlinkPipelineOptions.defaults(), numSplits)) {
      splitEnumerator.start();
      testContext.registerReader(0, "0");
      splitEnumerator.addReader(0);
      testContext.getExecutorService().triggerAll();

      List<FlinkSourceSplit<KV<Integer, Integer>>> splitsForReader =
          testContext.getSplitAssignments().get(0).getAssignedSplits();
      assertEquals(numSplits / numSubtasks, splitsForReader.size());

      splitEnumerator.addSplitsBack(splitsForReader, 0);
      splitEnumerator.addReader(0);
      assertEquals(2 * numSplits / numSubtasks, splitsForReader.size());
    }
  }