@Override
    public void close() throws IOException {
        // Consumer task will close the task, and it internally closes all the resources including the consumer.
        Utils.closeQuietly(consumerTask, "ConsumerTask");

        // Wait until the consumer thread finishes.
        try {
            consumerTaskThread.join();
        } catch (Exception e) {
            log.error("Encountered error while waiting for consumerTaskThread to finish.", e);
        }
    }
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