@Override
    public void handleSplitRequest(int subtask, @Nullable String hostname) {
        if (!context.registeredReaders().containsKey(subtask)) {
            // reader failed between sending the request and now. skip this request.
            return;
        }

        if (splitAssigner == null) {
            // No DynamicFilteringData is received before the first split request,
            // create a split assigner that handles all splits
            createSplitAssigner(null);
        }

        if (LOG.isDebugEnabled()) {
            final String hostInfo =
                    hostname == null ? "(no host locality info)" : "(on host '" + hostname + "')";
            LOG.debug("Subtask {} {} is requesting a file source split", subtask, hostInfo);
        }

        final Optional<FileSourceSplit> nextSplit = getNextUnassignedSplit(hostname);
        if (nextSplit.isPresent()) {
            final FileSourceSplit split = nextSplit.get();
            context.assignSplit((SplitT) split, subtask);
            assignedSplits.add(split.splitId());
            LOG.debug("Assigned split to subtask {} : {}", subtask, split);
        } else {
            context.signalNoMoreSplits(subtask);
            LOG.info("No more splits available for subtask {}", subtask);
        }
    }
@Test
    void testEnumerating() {
        String[] splits = new String[] {"0", "1", "2", "3", "4"};
        MockSplitEnumeratorContext<TestSplit> context = new MockSplitEnumeratorContext<>(1);
        context.registerReader(new ReaderInfo(0, ""));

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

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

        assertThat(getAssignedSplits(context)).containsExactlyInAnyOrder(splits);
    }