@Test public void testInitStartOffset() throws Exception { LOG.info("[testInitStartOffset] start"); properties.setProperty(TalosClientConfigKeys.GALAXY_TALOS_SERVICE_ENDPOINT, "testURI"); consumerConfig = new TalosConsumerConfig(properties, false); messageProcessorMock = Mockito.mock(MessageProcessor.class); messageReader = new GreedyMessageReader(consumerConfig); messageReader.setSimpleConsumer(simpleConsumerMock) .setConsumerClient(consumerClientMock) .setMessageProcessor(messageProcessorMock) .setTopicAndPartition(topicAndPartition) .setWorkerId(workerId); InOrder inOrder = inOrder(simpleConsumerMock, consumerClientMock, messageProcessorMock); doNothing().when(messageProcessorMock).init(topicAndPartition, startMessageOffset); messageReader.initStartOffset(); long startOffset = messageReader.getStartOffset().get(); assertEquals(startMessageOffset, startOffset); inOrder.verify(messageProcessorMock).init(topicAndPartition, startMessageOffset); doReturn(messageAndOffsetList).when(simpleConsumerMock).fetchMessage(startMessageOffset); doNothing().when(messageProcessorMock).process(messageAndOffsetList, messageReader); messageReader.fetchData(); inOrder.verify(simpleConsumerMock).fetchMessage(startMessageOffset); inOrder.verify(messageProcessorMock).process(messageAndOffsetList, messageReader); doNothing().when(messageProcessorMock).shutdown(messageReader); messageReader.commitCheckPoint(); inOrder.verify(messageProcessorMock).shutdown(messageReader); LOG.info("[testInitStartOffset] done"); }
@Override public void initStartOffset() { long readingStartOffset; if (outerCheckPoint != null && outerCheckPoint >= 0) { readingStartOffset = outerCheckPoint; outerCheckPoint = null; } else { readingStartOffset = queryStartOffset(); } if (readingStartOffset == -1 || consumerConfig.isResetOffsetWhenStart()) { startOffset.set(consumerConfig.getResetOffsetValueWhenStart()); } else { startOffset.set(readingStartOffset); } if (startOffset.longValue() > 0) { lastCommitOffset = finishedOffset = startOffset.get() - 1; } LOG.info("Init startOffset: " + startOffset + " lastCommitOffset: " + lastCommitOffset + " for partition: " + topicAndPartition); messageProcessor.init(topicAndPartition, startOffset.get()); }
GreedyMessageReader extends MessageReader implements MessageCheckpointer { @Override public void initStartOffset() { long readingStartOffset; if (outerCheckPoint != null && outerCheckPoint >= 0) { readingStartOffset = outerCheckPoint; outerCheckPoint = null; } else { readingStartOffset = queryStartOffset(); } if (readingStartOffset == -1 || consumerConfig.isResetOffsetWhenStart()) { startOffset.set(consumerConfig.getResetOffsetValueWhenStart()); } else { startOffset.set(readingStartOffset); } if (startOffset.longValue() > 0) { lastCommitOffset = finishedOffset = startOffset.get() - 1; } LOG.info("Init startOffset: " + startOffset + " lastCommitOffset: " + lastCommitOffset + " for partition: " + topicAndPartition); messageProcessor.init(topicAndPartition, startOffset.get()); } }
GreedyMessageReader extends MessageReader implements MessageCheckpointer { @Override public void initStartOffset() { long readingStartOffset; if (outerCheckPoint != null && outerCheckPoint >= 0) { readingStartOffset = outerCheckPoint; outerCheckPoint = null; } else { readingStartOffset = queryStartOffset(); } if (readingStartOffset == -1 || consumerConfig.isResetOffsetWhenStart()) { startOffset.set(consumerConfig.getResetOffsetValueWhenStart()); } else { startOffset.set(readingStartOffset); } if (startOffset.longValue() > 0) { lastCommitOffset = finishedOffset = startOffset.get() - 1; } LOG.info("Init startOffset: " + startOffset + " lastCommitOffset: " + lastCommitOffset + " for partition: " + topicAndPartition); messageProcessor.init(topicAndPartition, startOffset.get()); } GreedyMessageReader(TalosConsumerConfig talosConsumerConfig); }
GreedyMessageReader extends MessageReader implements MessageCheckpointer { @Override public void initStartOffset() { long readingStartOffset; if (outerCheckPoint != null && outerCheckPoint >= 0) { readingStartOffset = outerCheckPoint; outerCheckPoint = null; } else { readingStartOffset = queryStartOffset(); } if (readingStartOffset == -1 || consumerConfig.isResetOffsetWhenStart()) { startOffset.set(consumerConfig.getResetOffsetValueWhenStart()); } else { startOffset.set(readingStartOffset); } if (startOffset.longValue() > 0) { lastCommitOffset = finishedOffset = startOffset.get() - 1; } LOG.info("Init startOffset: " + startOffset + " lastCommitOffset: " + lastCommitOffset + " for partition: " + topicAndPartition); messageProcessor.init(topicAndPartition, startOffset.get()); } GreedyMessageReader(TalosConsumerConfig talosConsumerConfig); @Override void initStartOffset(); long queryStartOffset(); @Override void commitCheckPoint(); @Override void fetchData(); @Override void setOuterCheckPoint(Long outerCheckPoint); @Override boolean checkpoint(); @Override boolean checkpoint(long messageOffset); }
GreedyMessageReader extends MessageReader implements MessageCheckpointer { @Override public void initStartOffset() { long readingStartOffset; if (outerCheckPoint != null && outerCheckPoint >= 0) { readingStartOffset = outerCheckPoint; outerCheckPoint = null; } else { readingStartOffset = queryStartOffset(); } if (readingStartOffset == -1 || consumerConfig.isResetOffsetWhenStart()) { startOffset.set(consumerConfig.getResetOffsetValueWhenStart()); } else { startOffset.set(readingStartOffset); } if (startOffset.longValue() > 0) { lastCommitOffset = finishedOffset = startOffset.get() - 1; } LOG.info("Init startOffset: " + startOffset + " lastCommitOffset: " + lastCommitOffset + " for partition: " + topicAndPartition); messageProcessor.init(topicAndPartition, startOffset.get()); } GreedyMessageReader(TalosConsumerConfig talosConsumerConfig); @Override void initStartOffset(); long queryStartOffset(); @Override void commitCheckPoint(); @Override void fetchData(); @Override void setOuterCheckPoint(Long outerCheckPoint); @Override boolean checkpoint(); @Override boolean checkpoint(long messageOffset);  }