docs: public List<KafkaMessage> getListWithoutAck(Long timeout, TimeUnit unit, long offset) throws CanalClientException {
          waitClientRunning();
          if (!running) {
              return Lists.newArrayList();
          }
  
          if (offset > -1) {
              TopicPartition tp = new TopicPartition(topic, partition == null ? 0 : partition);
              kafkaConsumer.seek(tp, offset);
          }
  
          ConsumerRecords<String, Message> records = kafkaConsumer.poll(unit.toMillis(timeout));
  
          if (!records.isEmpty()) {
              List<KafkaMessage> messages = new ArrayList<>();
              for (ConsumerRecord<String, Message> record : records) {
                  KafkaMessage message = new KafkaMessage(record.value(), record.offset());
                  messages.add(message);
              }
              return messages;
          }
          return Lists.newArrayList();
      }