@Override
    public InputStatus pollNext(ReaderOutput<T> output) throws Exception {
        // make sure we have a fetch we are working on, or move to the next
        RecordsWithSplitIds<E> recordsWithSplitId = this.currentFetch;
        if (recordsWithSplitId == null) {
            recordsWithSplitId = getNextFetch(output);
            if (recordsWithSplitId == null) {
                return trace(finishedOrAvailableLater());
            }
        }

        // we need to loop here, because we may have to go across splits
        while (true) {
            // Process one record.
            final E record = recordsWithSplitId.nextRecordFromSplit();
            if (record != null) {
                // emit the record.
                numRecordsInCounter.inc(1);
                recordEmitter.emitRecord(record, currentSplitOutput, currentSplitContext.state);
                LOG.trace("Emitted record: {}", record);

                // We always emit MORE_AVAILABLE here, even though we do not strictly know whether
                // more is available. If nothing more is available, the next invocation will find
                // this out and return the correct status.
                // That means we emit the occasional 'false positive' for availability, but this
                // saves us doing checks for every record. Ultimately, this is cheaper.
                return trace(InputStatus.MORE_AVAILABLE);
            } else if (!moveToNextSplit(recordsWithSplitId, output)) {
                // The fetch is done and we just discovered that and have not emitted anything, yet.
                // We need to move to the next fetch. As a shortcut, we call pollNext() here again,
                // rather than emitting nothing and waiting for the caller to call us again.
                return pollNext(output);
            }
        }
    }
@Test
    void testRecordsWithSplitsRecycledWhenEmpty() throws Exception {
        final TestingRecordsWithSplitIds<String> records =
                new TestingRecordsWithSplitIds<>("test-split", "value1", "value2");
        final SourceReader<?, ?> reader = createReaderAndAwaitAvailable("test-split", records);

        // poll thrice: twice to get all records, one more to trigger recycle and moving to the next
        // split
        reader.pollNext(new TestingReaderOutput<>());
        reader.pollNext(new TestingReaderOutput<>());
        reader.pollNext(new TestingReaderOutput<>());

        assertThat(records.isRecycled()).isTrue();
    }