@Test public void testRemoveOffset() throws Exception { OffsetStore offsetStore = new RemoteBrokerOffsetStore(mQClientFactory, group); final MessageQueue messageQueue = new MessageQueue(topic, brokerName, 4); offsetStore.updateOffset(messageQueue, 1024, false); assertThat(offsetStore.readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY)).isEqualTo(1024); offsetStore.removeOffset(messageQueue); assertThat(offsetStore.readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY)).isEqualTo(-1); }
public void removeOffset(MessageQueue mq) { if (mq != null) { this.offsetTable.remove(mq); log.info("remove unnecessary messageQueue offset. group={}, mq={}, offsetTableSize={}", this.groupName, mq, offsetTable.size()); } }
RemoteBrokerOffsetStore implements OffsetStore { public void removeOffset(MessageQueue mq) { if (mq != null) { this.offsetTable.remove(mq); log.info("remove unnecessary messageQueue offset. group={}, mq={}, offsetTableSize={}", this.groupName, mq, offsetTable.size()); } } }
RemoteBrokerOffsetStore implements OffsetStore { public void removeOffset(MessageQueue mq) { if (mq != null) { this.offsetTable.remove(mq); log.info("remove unnecessary messageQueue offset. group={}, mq={}, offsetTableSize={}", this.groupName, mq, offsetTable.size()); } } RemoteBrokerOffsetStore(MQClientInstance mQClientFactory, String groupName); }
RemoteBrokerOffsetStore implements OffsetStore { public void removeOffset(MessageQueue mq) { if (mq != null) { this.offsetTable.remove(mq); log.info("remove unnecessary messageQueue offset. group={}, mq={}, offsetTableSize={}", this.groupName, mq, offsetTable.size()); } } RemoteBrokerOffsetStore(MQClientInstance mQClientFactory, String groupName); @Override void load(); @Override void updateOffset(MessageQueue mq, long offset, boolean increaseOnly); @Override long readOffset(final MessageQueue mq, final ReadOffsetType type); @Override void persistAll(Set<MessageQueue> mqs); @Override void persist(MessageQueue mq); void removeOffset(MessageQueue mq); @Override Map<MessageQueue, Long> cloneOffsetTable(String topic); @Override void updateConsumeOffsetToBroker(MessageQueue mq, long offset, boolean isOneway); }
RemoteBrokerOffsetStore implements OffsetStore { public void removeOffset(MessageQueue mq) { if (mq != null) { this.offsetTable.remove(mq); log.info("remove unnecessary messageQueue offset. group={}, mq={}, offsetTableSize={}", this.groupName, mq, offsetTable.size()); } } RemoteBrokerOffsetStore(MQClientInstance mQClientFactory, String groupName); @Override void load(); @Override void updateOffset(MessageQueue mq, long offset, boolean increaseOnly); @Override long readOffset(final MessageQueue mq, final ReadOffsetType type); @Override void persistAll(Set<MessageQueue> mqs); @Override void persist(MessageQueue mq); void removeOffset(MessageQueue mq); @Override Map<MessageQueue, Long> cloneOffsetTable(String topic); @Override void updateConsumeOffsetToBroker(MessageQueue mq, long offset, boolean isOneway);  }