@Test public void testCheckStartOffset() throws TException, IOException { Message message1 = new Message(ByteBuffer.wrap("message1".getBytes())); Message message2 = new Message(ByteBuffer.wrap("message2".getBytes())); Message message3 = new Message(ByteBuffer.wrap("message3".getBytes())); message1.setCreateTimestamp(System.currentTimeMillis()); message2.setCreateTimestamp(System.currentTimeMillis()); message3.setCreateTimestamp(System.currentTimeMillis()); message1.setMessageType(MessageType.BINARY); message2.setMessageType(MessageType.BINARY); message3.setMessageType(MessageType.BINARY); messageList.add(message1); messageList.add(message2); messageList.add(message3); MessageBlock messageBlock = Compression.compress(messageList, producerConfig.getCompressionType()); messageBlock.setStartMessageOffset(startOffset); List<MessageBlock> messageBlockList = new ArrayList<MessageBlock>(1); messageBlockList.add(messageBlock); MessageAndOffset messageAndOffset1 = new MessageAndOffset(message1, startOffset); MessageAndOffset messageAndOffset2 = new MessageAndOffset(message2, startOffset + 1); MessageAndOffset messageAndOffset3 = new MessageAndOffset(message3, startOffset + 2); messageAndOffsetList.add(messageAndOffset1); messageAndOffsetList.add(messageAndOffset2); messageAndOffsetList.add(messageAndOffset3); GetMessageResponse response = new GetMessageResponse(messageBlockList, 3, "testCheckStartOffset"); when(messageClientMock.getMessage(any(GetMessageRequest.class))) .thenReturn(response); when(scheduleInfoCacheMock.getOrCreateMessageClient(any(TopicAndPartition.class))) .thenReturn(messageClientMock); simpleConsumer.fetchMessage(MessageOffset.START_OFFSET.getValue(), 100); simpleConsumer.fetchMessage(MessageOffset.LATEST_OFFSET.getValue(), 100); simpleConsumer.fetchMessage(0, 100); simpleConsumer.fetchMessage(2, 100); simpleConsumer.fetchMessage(-1, 100); simpleConsumer.fetchMessage(-2, 100); try { simpleConsumer.fetchMessage(-3, 100); assertTrue("test check startOffset validity error", false); } catch (Exception e) { assertTrue(e instanceof IllegalArgumentException); } }
public List<MessageAndOffset> fetchMessage(long startOffset) throws TException, IOException { return fetchMessage(startOffset, consumerConfig.getMaxFetchRecords()); }
SimpleConsumer { public List<MessageAndOffset> fetchMessage(long startOffset) throws TException, IOException { return fetchMessage(startOffset, consumerConfig.getMaxFetchRecords()); } }
SimpleConsumer { public List<MessageAndOffset> fetchMessage(long startOffset) throws TException, IOException { return fetchMessage(startOffset, consumerConfig.getMaxFetchRecords()); } SimpleConsumer(TalosConsumerConfig consumerConfig,
      TopicAndPartition topicAndPartition, MessageService.Iface messageClient,
      String consumerIdPrefix);  SimpleConsumer(TalosConsumerConfig consumerConfig,
      TopicAndPartition topicAndPartition, MessageService.Iface messageClient);  SimpleConsumer(TalosConsumerConfig consumerConfig, String topicName,
      int partitionId, Credential credential); @Deprecated  SimpleConsumer(TalosConsumerConfig consumerConfig,
      TopicAndPartition topicAndPartition, Credential credential);  SimpleConsumer(TalosConsumerConfig consumerConfig,
      String topicName, int partitionId, TalosClientFactory talosClientFactory,
      String consumerIdPrefix);  SimpleConsumer(TalosConsumerConfig consumerConfig,
      TopicAndPartition topicAndPartition, TalosClientFactory talosClientFactory,
      MessageService.Iface messageClient, String consumerIdPrefix);  SimpleConsumer(TalosConsumerConfig consumerConfig,
      TopicAndPartition topicAndPartition, TalosClientFactory talosClientFactory);  SimpleConsumer(TalosConsumerConfig consumerConfig, String topicName,
      int partitionId, MessageService.Iface messageClient,
      TopicService.Iface topicClient,
      ScheduleInfoCache scheduleInfoCacheMock); }
SimpleConsumer { public List<MessageAndOffset> fetchMessage(long startOffset) throws TException, IOException { return fetchMessage(startOffset, consumerConfig.getMaxFetchRecords()); } SimpleConsumer(TalosConsumerConfig consumerConfig,
      TopicAndPartition topicAndPartition, MessageService.Iface messageClient,
      String consumerIdPrefix);  SimpleConsumer(TalosConsumerConfig consumerConfig,
      TopicAndPartition topicAndPartition, MessageService.Iface messageClient);  SimpleConsumer(TalosConsumerConfig consumerConfig, String topicName,
      int partitionId, Credential credential); @Deprecated  SimpleConsumer(TalosConsumerConfig consumerConfig,
      TopicAndPartition topicAndPartition, Credential credential);  SimpleConsumer(TalosConsumerConfig consumerConfig,
      String topicName, int partitionId, TalosClientFactory talosClientFactory,
      String consumerIdPrefix);  SimpleConsumer(TalosConsumerConfig consumerConfig,
      TopicAndPartition topicAndPartition, TalosClientFactory talosClientFactory,
      MessageService.Iface messageClient, String consumerIdPrefix);  SimpleConsumer(TalosConsumerConfig consumerConfig,
      TopicAndPartition topicAndPartition, TalosClientFactory talosClientFactory);  SimpleConsumer(TalosConsumerConfig consumerConfig, String topicName,
      int partitionId, MessageService.Iface messageClient,
      TopicService.Iface topicClient,
      ScheduleInfoCache scheduleInfoCacheMock); TopicTalosResourceName getTopicTalosResourceName(); void setSimpleConsumerId(String simpleConsumerId); String getSimpleConsumerId(); int getPartitionId(); TopicAndPartition getTopicAndPartition(); List<MessageAndOffset> fetchMessage(long startOffset); List<MessageAndOffset> fetchMessage(long startOffset,
      int maxFetchedNumber); void shutDown(); }
SimpleConsumer { public List<MessageAndOffset> fetchMessage(long startOffset) throws TException, IOException { return fetchMessage(startOffset, consumerConfig.getMaxFetchRecords()); } SimpleConsumer(TalosConsumerConfig consumerConfig,
      TopicAndPartition topicAndPartition, MessageService.Iface messageClient,
      String consumerIdPrefix);  SimpleConsumer(TalosConsumerConfig consumerConfig,
      TopicAndPartition topicAndPartition, MessageService.Iface messageClient);  SimpleConsumer(TalosConsumerConfig consumerConfig, String topicName,
      int partitionId, Credential credential); @Deprecated  SimpleConsumer(TalosConsumerConfig consumerConfig,
      TopicAndPartition topicAndPartition, Credential credential);  SimpleConsumer(TalosConsumerConfig consumerConfig,
      String topicName, int partitionId, TalosClientFactory talosClientFactory,
      String consumerIdPrefix);  SimpleConsumer(TalosConsumerConfig consumerConfig,
      TopicAndPartition topicAndPartition, TalosClientFactory talosClientFactory,
      MessageService.Iface messageClient, String consumerIdPrefix);  SimpleConsumer(TalosConsumerConfig consumerConfig,
      TopicAndPartition topicAndPartition, TalosClientFactory talosClientFactory);  SimpleConsumer(TalosConsumerConfig consumerConfig, String topicName,
      int partitionId, MessageService.Iface messageClient,
      TopicService.Iface topicClient,
      ScheduleInfoCache scheduleInfoCacheMock); TopicTalosResourceName getTopicTalosResourceName(); void setSimpleConsumerId(String simpleConsumerId); String getSimpleConsumerId(); int getPartitionId(); TopicAndPartition getTopicAndPartition(); List<MessageAndOffset> fetchMessage(long startOffset); List<MessageAndOffset> fetchMessage(long startOffset,
      int maxFetchedNumber); void shutDown();  }