@Test public void testNextPullOffset() throws Exception { MessageQueue messageQueue = new MessageQueue(); when(rocketmqPullConsume.fetchConsumeOffset(any(MessageQueue.class), anyBoolean())) .thenReturn(123L); assertThat(localMessageCache.nextPullOffset(new MessageQueue())).isEqualTo(123L); }
long nextPullOffset(MessageQueue remoteQueue) { if (!pullOffsetTable.containsKey(remoteQueue)) { try { pullOffsetTable.putIfAbsent(remoteQueue, rocketmqPullConsumer.fetchConsumeOffset(remoteQueue, false)); } catch (MQClientException e) { log.error("An error occurred in fetch consume offset process.", e); } } return pullOffsetTable.get(remoteQueue); }
LocalMessageCache implements ServiceLifecycle { long nextPullOffset(MessageQueue remoteQueue) { if (!pullOffsetTable.containsKey(remoteQueue)) { try { pullOffsetTable.putIfAbsent(remoteQueue, rocketmqPullConsumer.fetchConsumeOffset(remoteQueue, false)); } catch (MQClientException e) { log.error("An error occurred in fetch consume offset process.", e); } } return pullOffsetTable.get(remoteQueue); } }
LocalMessageCache implements ServiceLifecycle { long nextPullOffset(MessageQueue remoteQueue) { if (!pullOffsetTable.containsKey(remoteQueue)) { try { pullOffsetTable.putIfAbsent(remoteQueue, rocketmqPullConsumer.fetchConsumeOffset(remoteQueue, false)); } catch (MQClientException e) { log.error("An error occurred in fetch consume offset process.", e); } } return pullOffsetTable.get(remoteQueue); } LocalMessageCache(final DefaultMQPullConsumer rocketmqPullConsumer, final ClientConfig clientConfig); }
LocalMessageCache implements ServiceLifecycle { long nextPullOffset(MessageQueue remoteQueue) { if (!pullOffsetTable.containsKey(remoteQueue)) { try { pullOffsetTable.putIfAbsent(remoteQueue, rocketmqPullConsumer.fetchConsumeOffset(remoteQueue, false)); } catch (MQClientException e) { log.error("An error occurred in fetch consume offset process.", e); } } return pullOffsetTable.get(remoteQueue); } LocalMessageCache(final DefaultMQPullConsumer rocketmqPullConsumer, final ClientConfig clientConfig); @Override void startup(); @Override void shutdown(); }
LocalMessageCache implements ServiceLifecycle { long nextPullOffset(MessageQueue remoteQueue) { if (!pullOffsetTable.containsKey(remoteQueue)) { try { pullOffsetTable.putIfAbsent(remoteQueue, rocketmqPullConsumer.fetchConsumeOffset(remoteQueue, false)); } catch (MQClientException e) { log.error("An error occurred in fetch consume offset process.", e); } } return pullOffsetTable.get(remoteQueue); } LocalMessageCache(final DefaultMQPullConsumer rocketmqPullConsumer, final ClientConfig clientConfig); @Override void startup(); @Override void shutdown();  }