@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 testAbortDoesNotClearsResults() throws Exception {
        executeCallbackWithSyncWorker(
                (writer, worker) -> {
                    callStart(writer);
                    callAbort(writer);
                    worker.processAllRequests();
                    writer.getAndRemoveWriteResult(CHECKPOINT_ID);
                });
    }