docs: public List<KafkaFlatMessage> getFlatListWithoutAck(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);
              kafkaConsumer2.seek(tp, offset);
          }
  
          ConsumerRecords<String, String> records = kafkaConsumer2.poll(unit.toMillis(timeout));
          if (!records.isEmpty()) {
              List<KafkaFlatMessage> flatMessages = new ArrayList<>();
              for (ConsumerRecord<String, String> record : records) {
                  String flatMessageJson = record.value();
                  FlatMessage flatMessage = JSON.parseObject(flatMessageJson, FlatMessage.class);
                  KafkaFlatMessage message = new KafkaFlatMessage(flatMessage, record.offset());
                  flatMessages.add(message);
              }
  
              return flatMessages;
          }
          return Lists.newArrayList();
      }