@Override
    public void connectionEstablished(
            TieredStorageSubpartitionId subpartitionId,
            NettyConnectionWriter nettyConnectionWriter) {
        subpartitionProducerAgents[subpartitionId.getSubpartitionId()].connectionEstablished(
                nettyConnectionWriter);
        nettyConnectionEstablished[subpartitionId.getSubpartitionId()] = true;
    }
@Test
    void testRelease() {
        TieredStorageResourceRegistry resourceRegistry = new TieredStorageResourceRegistry();
        MemoryTierProducerAgent memoryTierProducerAgent =
                createMemoryTierProducerAgent(false, SEGMENT_SIZE_BYTES, resourceRegistry);

        AtomicBoolean isClosed = new AtomicBoolean(false);
        memoryTierProducerAgent.connectionEstablished(
                SUBPARTITION_ID,
                new TestingNettyConnectionWriter.Builder()
                        .setCloseFunction(
                                throwable -> {
                                    isClosed.set(true);
                                    return null;
                                })
                        .build());

        resourceRegistry.clearResourceFor(PARTITION_ID);
        assertThat(isClosed).isTrue();
    }