@Override
    public boolean tryStartNewSegment(
            TieredStorageSubpartitionId subpartitionId, int segmentId, int minNumBuffers) {
        boolean canStartNewSegment =
                nettyConnectionEstablished[subpartitionId.getSubpartitionId()]
                        // Ensure that a subpartition's memory tier does not excessively use
                        // buffers, which may result in insufficient buffers for other subpartitions
                        && subpartitionProducerAgents[subpartitionId.getSubpartitionId()]
                                        .numQueuedBuffers()
                                < subpartitionMaxQueuedBuffers
                        && (memoryManager.getMaxNonReclaimableBuffers(getMemoryTierName())
                                        - memoryManager.numOwnerRequestedBuffer(
                                                getMemoryTierName()))
                                > Math.max(numBuffersPerSegment, minNumBuffers)
                        && memoryManager.ensureCapacity(
                                Math.max(numBuffersPerSegment, minNumBuffers));
        if (canStartNewSegment) {
            subpartitionProducerAgents[subpartitionId.getSubpartitionId()].updateSegmentId(
                    segmentId);
        }
        return canStartNewSegment;
    }
@Test
    void testTryStartNewSegment() {
        try (MemoryTierProducerAgent memoryTierProducerAgent =
                createMemoryTierProducerAgent(false)) {
            assertThat(memoryTierProducerAgent.tryStartNewSegment(SUBPARTITION_ID, 0, 0)).isFalse();
            memoryTierProducerAgent.connectionEstablished(
                    SUBPARTITION_ID, new TestingNettyConnectionWriter.Builder().build());
            assertThat(memoryTierProducerAgent.tryStartNewSegment(SUBPARTITION_ID, 0, 0)).isTrue();
        }
    }