public void waitTillConsumptionCatchesUp(RecordMetadata recordMetadata,
                                             long timeoutMs) throws TimeoutException {
        int partition = recordMetadata.partition();
        // If the current assignment does not have the subscription for this partition then return immediately.
        if (!consumerTask.isMetadataPartitionAssigned(partition)) {
            throw new KafkaException("This consumer is not assigned to the target partition " + partition +
                    ". Currently assigned partitions: " + consumerTask.metadataPartitionsAssigned());
        }
        long offset = recordMetadata.offset();
        long startTimeMs = time.milliseconds();
        long consumeCheckIntervalMs = Math.min(CONSUME_RECHECK_INTERVAL_MS, timeoutMs);
        log.info("Wait until the consumer is caught up with the target partition {} up-to offset {}", partition, offset);
        while (true) {
            long readOffset = consumerTask.readOffsetForMetadataPartition(partition).orElse(-1L);
            if (readOffset >= offset) {
                return;
            }
            log.debug("Expected offset for partition {} is {}, but the read offset is {}. " +
                    "Sleeping for {} ms to retry again", partition, offset, readOffset, consumeCheckIntervalMs);
            if (time.milliseconds() - startTimeMs > timeoutMs) {
                log.warn("Expected offset for partition {} is {}, but the read offset is {}",
                        partition, offset, readOffset);
                throw new TimeoutException("Timed out in catching up with the expected offset by consumer.");
            }
            time.sleep(consumeCheckIntervalMs);
        }
    }
Waits if necessary for the consumption to reach the partition/offset of the given {@code RecordMetadata}

@param recordMetadata record metadata to be checked for consumption.
@param timeoutMs      wait timeout in milliseconds
@throws TimeoutException if this method execution did not complete with in the given {@code timeoutMs}.
storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerManager.java