@Test public void testComputePullFromWhereReturnedNotFound() throws Exception{ DefaultLitePullConsumer defaultLitePullConsumer = createStartLitePullConsumer(); defaultLitePullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); MessageQueue messageQueue = createMessageQueue(); when(offsetStore.readOffset(any(MessageQueue.class), any(ReadOffsetType.class))).thenReturn(-1L); long offset = rebalanceImpl.computePullFromWhere(messageQueue); assertThat(offset).isEqualTo(0); }
public void setConsumeFromWhere(ConsumeFromWhere consumeFromWhere) { if (consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET && consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET && consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_TIMESTAMP) { throw new RuntimeException("Invalid ConsumeFromWhere Value", null); } this.consumeFromWhere = consumeFromWhere; }
DefaultLitePullConsumer extends ClientConfig implements LitePullConsumer { public void setConsumeFromWhere(ConsumeFromWhere consumeFromWhere) { if (consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET && consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET && consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_TIMESTAMP) { throw new RuntimeException("Invalid ConsumeFromWhere Value", null); } this.consumeFromWhere = consumeFromWhere; } }
DefaultLitePullConsumer extends ClientConfig implements LitePullConsumer { public void setConsumeFromWhere(ConsumeFromWhere consumeFromWhere) { if (consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET && consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET && consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_TIMESTAMP) { throw new RuntimeException("Invalid ConsumeFromWhere Value", null); } this.consumeFromWhere = consumeFromWhere; } DefaultLitePullConsumer();  DefaultLitePullConsumer(final String consumerGroup);  DefaultLitePullConsumer(RPCHook rpcHook);  DefaultLitePullConsumer(final String consumerGroup, RPCHook rpcHook);  DefaultLitePullConsumer(final String namespace, final String consumerGroup, RPCHook rpcHook); }
DefaultLitePullConsumer extends ClientConfig implements LitePullConsumer { public void setConsumeFromWhere(ConsumeFromWhere consumeFromWhere) { if (consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET && consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET && consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_TIMESTAMP) { throw new RuntimeException("Invalid ConsumeFromWhere Value", null); } this.consumeFromWhere = consumeFromWhere; } DefaultLitePullConsumer();  DefaultLitePullConsumer(final String consumerGroup);  DefaultLitePullConsumer(RPCHook rpcHook);  DefaultLitePullConsumer(final String consumerGroup, RPCHook rpcHook);  DefaultLitePullConsumer(final String namespace, final String consumerGroup, RPCHook rpcHook); @Override void start(); @Override void shutdown(); @Override void subscribe(String topic, String subExpression); @Override void subscribe(String topic, MessageSelector messageSelector); @Override void unsubscribe(String topic); @Override void assign(Collection<MessageQueue> messageQueues); @Override List<MessageExt> poll(); @Override List<MessageExt> poll(long timeout); @Override void seek(MessageQueue messageQueue, long offset); @Override void pause(Collection<MessageQueue> messageQueues); @Override void resume(Collection<MessageQueue> messageQueues); @Override Collection<MessageQueue> fetchMessageQueues(String topic); @Override Long offsetForTimestamp(MessageQueue messageQueue, Long timestamp); @Override void registerTopicMessageQueueChangeListener(String topic,
        TopicMessageQueueChangeListener topicMessageQueueChangeListener); @Override void commitSync(); @Override Long committed(MessageQueue messageQueue); @Override void updateNameServerAddress(String nameServerAddress); @Override void seekToBegin(MessageQueue messageQueue); @Override void seekToEnd(MessageQueue messageQueue); @Override boolean isAutoCommit(); @Override void setAutoCommit(boolean autoCommit); int getPullThreadNums(); void setPullThreadNums(int pullThreadNums); long getAutoCommitIntervalMillis(); void setAutoCommitIntervalMillis(long autoCommitIntervalMillis); int getPullBatchSize(); void setPullBatchSize(int pullBatchSize); long getPullThresholdForAll(); void setPullThresholdForAll(long pullThresholdForAll); int getConsumeMaxSpan(); void setConsumeMaxSpan(int consumeMaxSpan); int getPullThresholdForQueue(); void setPullThresholdForQueue(int pullThresholdForQueue); int getPullThresholdSizeForQueue(); void setPullThresholdSizeForQueue(int pullThresholdSizeForQueue); AllocateMessageQueueStrategy getAllocateMessageQueueStrategy(); void setAllocateMessageQueueStrategy(AllocateMessageQueueStrategy allocateMessageQueueStrategy); long getBrokerSuspendMaxTimeMillis(); long getPollTimeoutMillis(); void setPollTimeoutMillis(long pollTimeoutMillis); OffsetStore getOffsetStore(); void setOffsetStore(OffsetStore offsetStore); boolean isUnitMode(); void setUnitMode(boolean isUnitMode); MessageModel getMessageModel(); void setMessageModel(MessageModel messageModel); String getConsumerGroup(); MessageQueueListener getMessageQueueListener(); void setMessageQueueListener(MessageQueueListener messageQueueListener); long getConsumerPullTimeoutMillis(); void setConsumerPullTimeoutMillis(long consumerPullTimeoutMillis); long getConsumerTimeoutMillisWhenSuspend(); void setConsumerTimeoutMillisWhenSuspend(long consumerTimeoutMillisWhenSuspend); long getTopicMetadataCheckIntervalMillis(); void setTopicMetadataCheckIntervalMillis(long topicMetadataCheckIntervalMillis); void setConsumerGroup(String consumerGroup); ConsumeFromWhere getConsumeFromWhere(); void setConsumeFromWhere(ConsumeFromWhere consumeFromWhere); String getConsumeTimestamp(); void setConsumeTimestamp(String consumeTimestamp); }
DefaultLitePullConsumer extends ClientConfig implements LitePullConsumer { public void setConsumeFromWhere(ConsumeFromWhere consumeFromWhere) { if (consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET && consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET && consumeFromWhere != ConsumeFromWhere.CONSUME_FROM_TIMESTAMP) { throw new RuntimeException("Invalid ConsumeFromWhere Value", null); } this.consumeFromWhere = consumeFromWhere; } DefaultLitePullConsumer();  DefaultLitePullConsumer(final String consumerGroup);  DefaultLitePullConsumer(RPCHook rpcHook);  DefaultLitePullConsumer(final String consumerGroup, RPCHook rpcHook);  DefaultLitePullConsumer(final String namespace, final String consumerGroup, RPCHook rpcHook); @Override void start(); @Override void shutdown(); @Override void subscribe(String topic, String subExpression); @Override void subscribe(String topic, MessageSelector messageSelector); @Override void unsubscribe(String topic); @Override void assign(Collection<MessageQueue> messageQueues); @Override List<MessageExt> poll(); @Override List<MessageExt> poll(long timeout); @Override void seek(MessageQueue messageQueue, long offset); @Override void pause(Collection<MessageQueue> messageQueues); @Override void resume(Collection<MessageQueue> messageQueues); @Override Collection<MessageQueue> fetchMessageQueues(String topic); @Override Long offsetForTimestamp(MessageQueue messageQueue, Long timestamp); @Override void registerTopicMessageQueueChangeListener(String topic,
        TopicMessageQueueChangeListener topicMessageQueueChangeListener); @Override void commitSync(); @Override Long committed(MessageQueue messageQueue); @Override void updateNameServerAddress(String nameServerAddress); @Override void seekToBegin(MessageQueue messageQueue); @Override void seekToEnd(MessageQueue messageQueue); @Override boolean isAutoCommit(); @Override void setAutoCommit(boolean autoCommit); int getPullThreadNums(); void setPullThreadNums(int pullThreadNums); long getAutoCommitIntervalMillis(); void setAutoCommitIntervalMillis(long autoCommitIntervalMillis); int getPullBatchSize(); void setPullBatchSize(int pullBatchSize); long getPullThresholdForAll(); void setPullThresholdForAll(long pullThresholdForAll); int getConsumeMaxSpan(); void setConsumeMaxSpan(int consumeMaxSpan); int getPullThresholdForQueue(); void setPullThresholdForQueue(int pullThresholdForQueue); int getPullThresholdSizeForQueue(); void setPullThresholdSizeForQueue(int pullThresholdSizeForQueue); AllocateMessageQueueStrategy getAllocateMessageQueueStrategy(); void setAllocateMessageQueueStrategy(AllocateMessageQueueStrategy allocateMessageQueueStrategy); long getBrokerSuspendMaxTimeMillis(); long getPollTimeoutMillis(); void setPollTimeoutMillis(long pollTimeoutMillis); OffsetStore getOffsetStore(); void setOffsetStore(OffsetStore offsetStore); boolean isUnitMode(); void setUnitMode(boolean isUnitMode); MessageModel getMessageModel(); void setMessageModel(MessageModel messageModel); String getConsumerGroup(); MessageQueueListener getMessageQueueListener(); void setMessageQueueListener(MessageQueueListener messageQueueListener); long getConsumerPullTimeoutMillis(); void setConsumerPullTimeoutMillis(long consumerPullTimeoutMillis); long getConsumerTimeoutMillisWhenSuspend(); void setConsumerTimeoutMillisWhenSuspend(long consumerTimeoutMillisWhenSuspend); long getTopicMetadataCheckIntervalMillis(); void setTopicMetadataCheckIntervalMillis(long topicMetadataCheckIntervalMillis); void setConsumerGroup(String consumerGroup); ConsumeFromWhere getConsumeFromWhere(); void setConsumeFromWhere(ConsumeFromWhere consumeFromWhere); String getConsumeTimestamp(); void setConsumeTimestamp(String consumeTimestamp);  }