@Test public void testFetchMessage() throws Exception { 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); long unHandledNumber = 117; GetMessageResponse response = new GetMessageResponse(messageBlockList, 3, "testFetchMessageSequenceId").setUnHandledMessageNumber(unHandledNumber); when(messageClientMock.getMessage(any(GetMessageRequest.class))) .thenReturn(response); when(scheduleInfoCacheMock.getOrCreateMessageClient(any(TopicAndPartition.class))) .thenReturn(messageClientMock); List<MessageAndOffset> msgList = simpleConsumer.fetchMessage(startOffset); for (int i = 0; i < msgList.size(); ++i) { assertEquals(messageAndOffsetList.get(i).getMessage(), msgList.get(i).getMessage()); assertEquals(messageAndOffsetList.get(i).getMessageOffset(), msgList.get(i).getMessageOffset()); assertEquals(msgList.get(i).getUnHandledMessageNumber(), unHandledNumber + msgList.size() - 1 - i); } try { simpleConsumer.fetchMessage(startOffset, 3000); assertTrue("test fetchMessage illegal maxFetchNumber error", false); } catch (Exception e) { assertTrue(e instanceof IllegalArgumentException); } InOrder inOrder = inOrder(messageClientMock); inOrder.verify(messageClientMock).getMessage(any(GetMessageRequest.class)); }
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();  }