docs: @Override
      public Message get(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit)
                                                                                                   throws CanalServerException {
          checkStart(clientIdentity.getDestination());
          checkSubscribe(clientIdentity);
          CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
          synchronized (canalInstance) {
              // 获取到流式数据中的最后一批获取的位置
              PositionRange<LogPosition> positionRanges = canalInstance.getMetaManager().getLastestBatch(clientIdentity);
  
              if (positionRanges != null) {
                  throw new CanalServerException(String.format("clientId:%s has last batch:[%s] isn't ack , maybe loss data",
                      clientIdentity.getClientId(),
                      positionRanges));
              }
  
              Events<Event> events = null;
              Position start = canalInstance.getMetaManager().getCursor(clientIdentity);
              events = getEvents(canalInstance.getEventStore(), start, batchSize, timeout, unit);
  
              if (CollectionUtils.isEmpty(events.getEvents())) {
                  logger.debug("get successfully, clientId:{} batchSize:{} but result is null",
                      clientIdentity.getClientId(),
                      batchSize);
                  return new Message(-1, true, new ArrayList()); // 返回空包，避免生成batchId，浪费性能
              } else {
                  // 记录到流式信息
                  Long batchId = canalInstance.getMetaManager().addBatch(clientIdentity, events.getPositionRange());
                  boolean raw = isRaw(canalInstance.getEventStore());
                  List entrys = null;
                  if (raw) {
                      entrys = Lists.transform(events.getEvents(), new Function<Event, ByteString>() {
  
                          public ByteString apply(Event input) {
                              return input.getRawEntry();
                          }
                      });
                  } else {
                      entrys = Lists.transform(events.getEvents(), new Function<Event, CanalEntry.Entry>() {
  
                          public CanalEntry.Entry apply(Event input) {
                              return input.getEntry();
                          }
                      });
                  }
                  if (logger.isInfoEnabled()) {
                      logger.info("get successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]",
                          clientIdentity.getClientId(),
                          batchSize,
                          entrys.size(),
                          batchId,
                          events.getPositionRange());
                  }
                  // 直接提交ack
                  ack(clientIdentity, batchId);
                  return new Message(batchId, raw, entrys);
              }
          }
      }