@Override
    public void close() {
        // close is called when task is finished or failed.
        checkNotNull(memoryDataManager).close();
        super.close();
    }
@Test
    void testClose() throws Exception {
        final int numBuffers = 1;

        BufferPool bufferPool = globalPool.createBufferPool(numBuffers, numBuffers);
        HsResultPartition partition = createHsResultPartition(1, bufferPool);

        partition.close();
        // emit data to closed partition will throw exception.
        assertThatThrownBy(() -> partition.emitRecord(ByteBuffer.allocate(bufferSize), 0));
    }