@Override
    public void close() throws IOException {
        if (wasClosed.compareAndSet(false, true)) {
            LOG.debug("close, dropping checkpoints {}", results.keySet());
            results.clear();
            enqueue(ChannelStateWriteRequest.releaseSubtask(jobVertexID, subtaskIndex), false);
            executor.releaseSubtask(jobVertexID, subtaskIndex);
        }
    }
@Test
    void testNoAddDataAfterClose() throws IOException {
        ChannelStateWriterImpl writer = openWriter();
        callStart(writer);
        writer.close();
        assertThatThrownBy(() -> callAddInputData(writer))
                .hasCauseInstanceOf(IllegalStateException.class);
    }