@Override
    public ChannelStateWriteResult getAndRemoveWriteResult(long checkpointId) {
        LOG.debug("{} requested write result, checkpoint {}", taskName, checkpointId);
        ChannelStateWriteResult result = results.remove(checkpointId);
        Preconditions.checkArgument(
                result != null,
                taskName + " channel state write result not found for checkpoint " + checkpointId);
        return result;
    }
@Test
    void testResultCompletion() throws IOException {
        ChannelStateWriteResult result;
        try (ChannelStateWriterImpl writer = openWriter()) {
            callStart(writer);
            result = writer.getAndRemoveWriteResult(CHECKPOINT_ID);
            assertThat(result.resultSubpartitionStateHandles).isNotDone();
            assertThat(result.inputChannelStateHandles).isNotDone();
        }
        assertThat(result.inputChannelStateHandles).isDone();
        assertThat(result.resultSubpartitionStateHandles).isDone();
    }