@Override
    public CompletableFuture<Void> prepareSnapshot(
            ChannelStateWriter channelStateWriter, long checkpointId) throws CheckpointException {
        for (Map.Entry<
                        InputChannelInfo,
                        SpillingAdaptiveSpanningRecordDeserializer<
                                DeserializationDelegate<StreamElement>>>
                e : recordDeserializers.entrySet()) {

            try {
                channelStateWriter.addInputData(
                        checkpointId,
                        e.getKey(),
                        ChannelStateWriter.SEQUENCE_NUMBER_UNKNOWN,
                        e.getValue().getUnconsumedBuffer());
            } catch (IOException ioException) {
                throw new CheckpointException(CheckpointFailureReason.IO_EXCEPTION, ioException);
            }
        }
        return checkpointedInputGate.getAllBarriersReceivedFuture(checkpointId);
    }
@Test
    void testSnapshotAfterEndOfPartition() throws Exception {
        int numInputChannels = 1;
        int channelId = 0;
        int checkpointId = 0;

        VerifyRecordsDataOutput<Long> output = new VerifyRecordsDataOutput<>();
        LongSerializer inSerializer = LongSerializer.INSTANCE;
        StreamTestSingleInputGate<Long> inputGate =
                new StreamTestSingleInputGate<>(numInputChannels, 0, inSerializer, 1024);
        StreamTaskInput<Long> input =
                new StreamTaskNetworkInput<>(
                        new CheckpointedInputGate(
                                inputGate.getInputGate(),
                                SingleCheckpointBarrierHandler
                                        .createUnalignedCheckpointBarrierHandler(
                                                TestSubtaskCheckpointCoordinator.INSTANCE,
                                                "test",
                                                new DummyCheckpointInvokable(),
                                                SystemClock.getInstance(),
                                                false,
                                                inputGate.getInputGate()),
                                new SyncMailboxExecutor()),
                        inSerializer,
                        ioManager,
                        new StatusWatermarkValve(numInputChannels),
                        0,
                        () -> false);

        inputGate.sendEvent(
                new CheckpointBarrier(
                        checkpointId,
                        0L,
                        CheckpointOptions.forCheckpointWithDefaultLocation().toUnaligned()),
                channelId);
        inputGate.sendElement(new StreamRecord<>(42L), channelId);

        assertHasNextElement(input, output);
        assertHasNextElement(input, output);
        assertThat(output.getNumberOfEmittedRecords()).isOne();

        // send EndOfPartitionEvent and ensure that deserializer has been released
        inputGate.sendEvent(EndOfPartitionEvent.INSTANCE, channelId);
        input.emitNext(output);

        // now snapshot all inflight buffers
        CompletableFuture<Void> completableFuture =
                input.prepareSnapshot(ChannelStateWriter.NO_OP, checkpointId);
        completableFuture.join();
    }