@Override
    public void addSplitsBack(List<SplitT> splits, int subtaskId) {
        LOG.debug("Dynamic File Source Enumerator adds splits back: {}", splits);
        if (splitAssigner != null) {
            List<FileSourceSplit> fileSplits = new ArrayList<>(splits);
            // Only add back splits enumerating. A split may be filtered after it is assigned.
            fileSplits.removeIf(s -> !allEnumeratingSplits.contains(s.splitId()));
            // Added splits should be removed from assignedSplits for re-assignment
            fileSplits.forEach(s -> assignedSplits.remove(s.splitId()));
            splitAssigner.addSplits(fileSplits);
        }
    }
@Test
    void testAddSplitsBack() {
        String[] splits = new String[] {"0", "1", "2", "3", "4"};
        String[] remainingSplits = new String[] {"1", "3"};
        MockSplitEnumeratorContext<TestSplit> context = new MockSplitEnumeratorContext<>(1);
        context.registerReader(new ReaderInfo(0, ""));

        DynamicFileSplitEnumerator<TestSplit> enumerator =
                new DynamicFileSplitEnumerator<>(
                        context,
                        () -> new TestDynamicFileEnumerator(splits, remainingSplits),
                        SimpleSplitAssigner::new);

        for (String ignored : splits) {
            enumerator.handleSplitRequest(0, null);
        }

        enumerator.handleSourceEvent(0, mockDynamicFilteringEvent());

        enumerator.addSplitsBack(
                Arrays.stream(splits).map(TestSplit::new).collect(Collectors.toList()), 0);

        for (String ignored : splits) {
            enumerator.handleSplitRequest(0, null);
        }

        List<String> assignedSplits = getAssignedSplits(context);
        assertThat(assignedSplits.subList(5, assignedSplits.size()))
                .containsExactlyInAnyOrder(remainingSplits);
    }