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