@Override
    public void updateAndFreezeOffset(MessageQueue mq, long offset) {
        if (mq != null) {
            this.offsetTable.computeIfAbsent(mq, k -> new ControllableOffset(offset))
                .updateAndFreeze(offset);
        }
    }
@Test
    public void testUpdateAndFreezeOffset() throws Exception {
        OffsetStore offsetStore = new RemoteBrokerOffsetStore(mQClientFactory, group);
        MessageQueue messageQueue = new MessageQueue(topic, brokerName, 1);

        offsetStore.updateAndFreezeOffset(messageQueue, 1024);
        assertThat(offsetStore.readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY)).isEqualTo(1024);

        offsetStore.updateOffset(messageQueue, 1023, false);
        assertThat(offsetStore.readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY)).isEqualTo(1024);

        offsetStore.updateOffset(messageQueue, 1022, true);
        assertThat(offsetStore.readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY)).isEqualTo(1024);
    }