### Question:
MessageServiceImpl implements MessageService { @Override public List<MessageView> queryMessageByTopic(String topic, final long begin, final long end) { DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(MixAll.TOOLS_CONSUMER_GROUP, null); List<MessageView> messageViewList = Lists.newArrayList(); try { String subExpression = "*"; consumer.start(); Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues(topic); for (MessageQueue mq : mqs) { long minOffset = consumer.searchOffset(mq, begin); long maxOffset = consumer.searchOffset(mq, end); READQ: for (long offset = minOffset; offset <= maxOffset; ) { try { if (messageViewList.size() > 2000) { break; } PullResult pullResult = consumer.pull(mq, subExpression, offset, 32); offset = pullResult.getNextBeginOffset(); switch (pullResult.getPullStatus()) { case FOUND: List<MessageView> messageViewListByQuery = Lists.transform(pullResult.getMsgFoundList(), new Function<MessageExt, MessageView>() { @Override public MessageView apply(MessageExt messageExt) { messageExt.setBody(null); return MessageView.fromMessageExt(messageExt); } }); List<MessageView> filteredList = Lists.newArrayList(Iterables.filter(messageViewListByQuery, new Predicate<MessageView>() { @Override public boolean apply(MessageView messageView) { if (messageView.getStoreTimestamp() < begin || messageView.getStoreTimestamp() > end) { logger.info("begin={} end={} time not in range {} {}", begin, end, messageView.getStoreTimestamp(), new Date(messageView.getStoreTimestamp()).toString()); } return messageView.getStoreTimestamp() >= begin && messageView.getStoreTimestamp() <= end; } })); messageViewList.addAll(filteredList); break; case NO_MATCHED_MSG: case NO_NEW_MSG: case OFFSET_ILLEGAL: break READQ; } } catch (Exception e) { break; } } } Collections.sort(messageViewList, new Comparator<MessageView>() { @Override public int compare(MessageView o1, MessageView o2) { if (o1.getStoreTimestamp() - o2.getStoreTimestamp() == 0) { return 0; } return (o1.getStoreTimestamp() > o2.getStoreTimestamp()) ? -1 : 1; } }); return messageViewList; } catch (Exception e) { throw Throwables.propagate(e); } finally { consumer.shutdown(); } }  Pair<MessageView, List<MessageTrack>> viewMessage(String subject, final String msgId); @Override List<MessageView> queryMessageByTopicAndKey(String topic, String key); @Override List<MessageView> queryMessageByTopic(String topic, final long begin, final long end); @Override List<MessageTrack> messageTrackDetail(MessageExt msg); @Override ConsumeMessageDirectlyResult consumeMessageDirectly(String topic, String msgId, String consumerGroup,
        String clientId);  }### Answer:
@Test public void queryMessageByTopic() throws Exception { final String messageId = msgId; List<MessageView> messageViewList = null; for (int i = 0; i < 10; i++) { messageViewList = messageService.queryMessageByTopic(TEST_CONSOLE_TOPIC, System.currentTimeMillis() - 100000, System.currentTimeMillis()); if (CollectionUtils.isEmpty(messageViewList)) { Thread.sleep(1000); continue; } break; } Assert.assertTrue(CollectionUtils.isNotEmpty(messageViewList)); Assert.assertTrue(Lists.transform(messageViewList, new Function<MessageView, String>() { @Override public String apply(MessageView input) { return input.getMsgId(); } }).contains(messageId)); }