@Test public void testUpdateOffset() throws Exception { OffsetStore offsetStore = new RemoteBrokerOffsetStore(mQClientFactory, group); MessageQueue messageQueue = new MessageQueue(topic, brokerName, 1); offsetStore.updateOffset(messageQueue, 1024, false); assertThat(offsetStore.readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY)).isEqualTo(1024); offsetStore.updateOffset(messageQueue, 1023, false); assertThat(offsetStore.readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY)).isEqualTo(1023); offsetStore.updateOffset(messageQueue, 1022, true); assertThat(offsetStore.readOffset(messageQueue, ReadOffsetType.READ_FROM_MEMORY)).isEqualTo(1023); }
@Override public void updateOffset(MessageQueue mq, long offset, boolean increaseOnly) { if (mq != null) { AtomicLong offsetOld = this.offsetTable.get(mq); if (null == offsetOld) { offsetOld = this.offsetTable.putIfAbsent(mq, new AtomicLong(offset)); } if (null != offsetOld) { if (increaseOnly) { MixAll.compareAndIncreaseOnly(offsetOld, offset); } else { offsetOld.set(offset); } } } }
RemoteBrokerOffsetStore implements OffsetStore { @Override public void updateOffset(MessageQueue mq, long offset, boolean increaseOnly) { if (mq != null) { AtomicLong offsetOld = this.offsetTable.get(mq); if (null == offsetOld) { offsetOld = this.offsetTable.putIfAbsent(mq, new AtomicLong(offset)); } if (null != offsetOld) { if (increaseOnly) { MixAll.compareAndIncreaseOnly(offsetOld, offset); } else { offsetOld.set(offset); } } } } }
RemoteBrokerOffsetStore implements OffsetStore { @Override public void updateOffset(MessageQueue mq, long offset, boolean increaseOnly) { if (mq != null) { AtomicLong offsetOld = this.offsetTable.get(mq); if (null == offsetOld) { offsetOld = this.offsetTable.putIfAbsent(mq, new AtomicLong(offset)); } if (null != offsetOld) { if (increaseOnly) { MixAll.compareAndIncreaseOnly(offsetOld, offset); } else { offsetOld.set(offset); } } } } RemoteBrokerOffsetStore(MQClientInstance mQClientFactory, String groupName); }
RemoteBrokerOffsetStore implements OffsetStore { @Override public void updateOffset(MessageQueue mq, long offset, boolean increaseOnly) { if (mq != null) { AtomicLong offsetOld = this.offsetTable.get(mq); if (null == offsetOld) { offsetOld = this.offsetTable.putIfAbsent(mq, new AtomicLong(offset)); } if (null != offsetOld) { if (increaseOnly) { MixAll.compareAndIncreaseOnly(offsetOld, offset); } else { offsetOld.set(offset); } } } } 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 { @Override public void updateOffset(MessageQueue mq, long offset, boolean increaseOnly) { if (mq != null) { AtomicLong offsetOld = this.offsetTable.get(mq); if (null == offsetOld) { offsetOld = this.offsetTable.putIfAbsent(mq, new AtomicLong(offset)); } if (null != offsetOld) { if (increaseOnly) { MixAll.compareAndIncreaseOnly(offsetOld, offset); } else { offsetOld.set(offset); } } } } 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);  }