@SuppressWarnings("unchecked")
    @Override
    public List<CommonMessage> getMessage(Long timeout, TimeUnit unit) {
        List<CommonMessage> messageList = Lists.newArrayList();
        try {
            Messages<byte[]> messages = pulsarMQConsumer.batchReceive();
            if (null == messages || messages.size() == 0) {
                return messageList;
            }
            // 保存当前消费记录，用于ack和rollback
            this.lastGetBatchMessage = messages;
            for (org.apache.pulsar.client.api.Message<byte[]> msg : messages) {
                byte[] data = msg.getData();
                if (!this.flatMessage) {
                    Message message = CanalMessageSerializerUtil.deserializer(data);
                    List<CommonMessage> list = MessageUtil.convert(message);
                    messageList.addAll(list);
                } else {
                    CommonMessage commonMessage = JSON.parseObject(data, CommonMessage.class);
                    messageList.add(commonMessage);
                }
            }
        } catch (PulsarClientException e) {
            throw new CanalClientException("Receive pulsar batch message error", e);
        }

        return messageList;
    }
@Test
    public void getMessage() {
        consumer.connect();

        while (true) {
            List<CommonMessage> list = consumer.getMessage(-1L, TimeUnit.SECONDS);
            String time = LocalTime.now().format(DateTimeFormatter.ofPattern("HH:mm:ss"));
            if (null == list || list.isEmpty()) {
                System.out.println(time + " Receive empty");
                continue;
            }
            for (CommonMessage m : list) {
                System.out.println(time + " Receive ==> " + JSON.toJSONString(m));
            }

            consumer.ack();
        }
    }