@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));
}