public List<ConsumerRecord<String, String>> consume(TopicPartition topicPartition,
                                                        Integer expectedTotalCount,
                                                        Long fetchOffset) {
        consumer.assign(Collections.singletonList(topicPartition));
        consumer.seek(topicPartition, fetchOffset);

        long timeoutMs = 60_000L;
        long pollTimeoutMs = 100L;
        String sep = System.lineSeparator();
        List<ConsumerRecord<String, String>> records = new ArrayList<>();
        Function1<ConsumerRecords<String, String>, Object> pollAction = polledRecords -> {
            polledRecords.forEach(records::add);
            return records.size() >= expectedTotalCount;
        };
        Function0<String> messageSupplier = () ->
                String.format("Could not consume %d records of %s from offset %d in %d ms. %d message(s) consumed:%s%s",
                        expectedTotalCount, topicPartition, fetchOffset, timeoutMs, records.size(), sep,
                        records.stream().map(Object::toString).collect(Collectors.joining(sep)));
        TestUtils.pollRecordsUntilTrue(consumer, pollAction, messageSupplier, timeoutMs, pollTimeoutMs);
        return records;
    }
Send the given records trying to honor the batch size. This is attempted
with a large producer linger and the use of an explicit flush every time
the number of a "group" of records reaches the batch size.
@param recordsToProduce the records to produce
@param batchSize the batch size
storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestContext.java