@SuppressWarnings("FieldAccessNotGuarded")
    public HsSubpartitionConsumerMemoryDataManager registerNewConsumer(HsConsumerId consumerId) {
        return callWithLock(
                () -> {
                    checkState(!consumerMap.containsKey(consumerId));
                    HsSubpartitionConsumerMemoryDataManager newConsumer =
                            new HsSubpartitionConsumerMemoryDataManager(
                                    resultPartitionLock,
                                    subpartitionLock.readLock(),
                                    targetChannel,
                                    consumerId,
                                    memoryDataManagerOperation);
                    newConsumer.addInitialBuffers(allBuffers);
                    consumerMap.put(consumerId, newConsumer);
                    return newConsumer;
                });
    }
@Test
    void testConsumerRegisterRepeatedly() {
        TestingMemoryDataManagerOperation memoryDataManagerOperation =
                TestingMemoryDataManagerOperation.builder().build();
        HsSubpartitionMemoryDataManager subpartitionMemoryDataManager =
                createSubpartitionMemoryDataManager(memoryDataManagerOperation);

        HsConsumerId consumerId = HsConsumerId.newId(null);
        subpartitionMemoryDataManager.registerNewConsumer(consumerId);
        assertThatThrownBy(() -> subpartitionMemoryDataManager.registerNewConsumer(consumerId))
                .isInstanceOf(IllegalStateException.class);
    }