public static List<ConsumerRecord<byte[], byte[]>> drainAllRecordsFromTopic(
      String topic, Properties properties) throws KafkaException {
    final Properties consumerConfig = new Properties();
    consumerConfig.putAll(properties);
    consumerConfig.put("key.deserializer", ByteArrayDeserializer.class.getName());
    consumerConfig.put("value.deserializer", ByteArrayDeserializer.class.getName());
    try (KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerConfig)) {
      Set<TopicPartition> topicPartitions = getAllPartitions(consumer, topic);
      Map<TopicPartition, Long> endOffsets = consumer.endOffsets(topicPartitions);
      consumer.assign(topicPartitions);
      consumer.seekToBeginning(topicPartitions);

      final List<ConsumerRecord<byte[], byte[]>> consumerRecords = new ArrayList<>();
      while (!topicPartitions.isEmpty()) {
        ConsumerRecords<byte[], byte[]> records = consumer.poll(CONSUMER_POLL_DURATION);
        LOG.debug("Fetched {} records from topic {}.", records.count(), topic);

        // Remove partitions from polling which have reached its end.
        final List<TopicPartition> finishedPartitions = new ArrayList<>();
        for (final TopicPartition topicPartition : topicPartitions) {
          final long position = consumer.position(topicPartition);
          final long endOffset = endOffsets.get(topicPartition);
          LOG.debug(
              "Endoffset {} and current position {} for partition {}",
              endOffset,
              position,
              topicPartition.partition());
          if (endOffset - position > 0) {
            continue;
          }
          finishedPartitions.add(topicPartition);
        }
        if (topicPartitions.removeAll(finishedPartitions)) {
          consumer.assign(topicPartitions);
        }
        for (ConsumerRecord<byte[], byte[]> r : records) {
          consumerRecords.add(r);
        }
      }
      return consumerRecords;
    }
  }
Drain all records available from the given topic from the beginning until the current highest
offset.

<p>This method will fetch the latest offsets for the partitions once and only return records
until that point.

@param topic to fetch from
@param properties used to configure the created {@link KafkaConsumer}
@return all {@link ConsumerRecord} in the topic
@throws KafkaException
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/kafka/testutils/KafkaUtil.java