@Override
    public DataInputStatus emitNext(DataOutput<T> output) throws Exception {
        if (sortedInput != null) {
            return emitNextSortedRecord(output);
        }

        DataInputStatus inputStatus = wrappedInput.emitNext(forwardingDataOutput);
        if (inputStatus == DataInputStatus.END_OF_DATA) {
            endSorting();
            return emitNextSortedRecord(output);
        }

        return inputStatus;
    }
@Test
    void simpleFixedLengthKeySorting() throws Exception {
        CollectingDataOutput<Integer> collectingDataOutput = new CollectingDataOutput<>();
        CollectionDataInput<Integer> input =
                new CollectionDataInput<>(
                        Arrays.asList(
                                new StreamRecord<>(1, 3),
                                new StreamRecord<>(1, 1),
                                new StreamRecord<>(2, 1),
                                new StreamRecord<>(2, 3),
                                new StreamRecord<>(1, 2),
                                new StreamRecord<>(2, 2)));
        MockEnvironment environment = MockEnvironment.builder().build();
        SortingDataInput<Integer, Integer> sortingDataInput =
                new SortingDataInput<>(
                        input,
                        new IntSerializer(),
                        new IntSerializer(),
                        (KeySelector<Integer, Integer>) value -> value,
                        environment.getMemoryManager(),
                        environment.getIOManager(),
                        true,
                        1.0,
                        new Configuration(),
                        new DummyInvokable(),
                        new ExecutionConfig());

        DataInputStatus inputStatus;
        do {
            inputStatus = sortingDataInput.emitNext(collectingDataOutput);
        } while (inputStatus != DataInputStatus.END_OF_INPUT);

        assertThat(collectingDataOutput.events)
                .containsExactly(
                        new StreamRecord<>(1, 1),
                        new StreamRecord<>(1, 2),
                        new StreamRecord<>(1, 3),
                        new StreamRecord<>(2, 1),
                        new StreamRecord<>(2, 2),
                        new StreamRecord<>(2, 3));
    }