@Override
    public void abort(long checkpointId, Throwable cause, boolean cleanup) {
        LOG.debug("{} aborting, checkpoint {}", taskName, checkpointId);
        enqueue(
                ChannelStateWriteRequest.abort(jobVertexID, subtaskIndex, checkpointId, cause),
                true); // abort already started
        enqueue(
                ChannelStateWriteRequest.abort(jobVertexID, subtaskIndex, checkpointId, cause),
                false); // abort enqueued but not started
        if (cleanup) {
            results.remove(checkpointId);
        }
    }
@Test
    void testAbort() throws Exception {
        NetworkBuffer buffer = getBuffer();
        executeCallbackWithSyncWorker(
                (writer, worker) -> {
                    callStart(writer);
                    ChannelStateWriteResult result = writer.getAndRemoveWriteResult(CHECKPOINT_ID);
                    callAddInputData(writer, buffer);
                    callAbort(writer);
                    worker.processAllRequests();
                    assertThat(result.isDone()).isTrue();
                    assertThat(buffer.isRecycled()).isTrue();
                });
    }