content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.clients.consumer;
  
  import org.apache.kafka.clients.KafkaClient;
  import org.apache.kafka.clients.consumer.internals.ConsumerDelegate;
  import org.apache.kafka.clients.consumer.internals.ConsumerDelegateCreator;
  import org.apache.kafka.clients.consumer.internals.ConsumerMetadata;
  import org.apache.kafka.clients.consumer.internals.SubscriptionState;
  import org.apache.kafka.clients.consumer.internals.metrics.KafkaConsumerMetrics;
  import org.apache.kafka.common.KafkaException;
  import org.apache.kafka.common.Metric;
  import org.apache.kafka.common.MetricName;
  import org.apache.kafka.common.PartitionInfo;
  import org.apache.kafka.common.TopicPartition;
  import org.apache.kafka.common.Uuid;
  import org.apache.kafka.common.errors.InterruptException;
  import org.apache.kafka.common.errors.InvalidRegularExpression;
  import org.apache.kafka.common.metrics.KafkaMetric;
  import org.apache.kafka.common.metrics.Metrics;
  import org.apache.kafka.common.serialization.Deserializer;
  import org.apache.kafka.common.utils.LogContext;
  import org.apache.kafka.common.utils.Time;
  import org.apache.kafka.common.utils.Timer;
  
  import java.time.Duration;
  import java.util.Collection;
  import java.util.ConcurrentModificationException;
  import java.util.List;
  import java.util.Map;
  import java.util.OptionalLong;
  import java.util.Properties;
  import java.util.Set;
  import java.util.regex.Pattern;
  
  import static org.apache.kafka.common.utils.Utils.propsToMap;
  
  /**
   * A client that consumes records from a Kafka cluster.
   * <p>
   * This client transparently handles the failure of Kafka brokers, and transparently adapts as topic partitions
   * it fetches migrate within the cluster. This client also interacts with the broker to allow groups of
   * consumers to load balance consumption using <a href="#consumergroups">consumer groups</a>.
   * <p>
   * The consumer maintains TCP connections to the necessary brokers to fetch data.
   * Failure to close the consumer after use will leak these connections.
   * The consumer is not thread-safe. See <a href="#multithreaded">Multi-threaded Processing</a> for more details.
   *
   * <h3>Cross-Version Compatibility</h3>
   * This client can communicate with brokers that are version 0.10.0 or newer. Older or newer brokers may not support
   * certain features. For example, 0.10.0 brokers do not support offsetsForTimes, because this feature was added
   * in version 0.10.1. You will receive an {@link org.apache.kafka.common.errors.UnsupportedVersionException}
   * when invoking an API that is not available on the running broker version.
   * <p>
   *
   * <h3>Offsets and Consumer Position</h3>
   * Kafka maintains a numerical offset for each record in a partition. This offset acts as a unique identifier of
   * a record within that partition, and also denotes the position of the consumer in the partition. For example, a consumer
   * which is at position 5 has consumed records with offsets 0 through 4 and will next receive the record with offset 5.
   * Note that offsets are not guaranteed to be consecutive (such as compacted topic or when records have been produced
   * using transactions). For example, if the consumer did read a record with offset 4, but 5 is not an offset
   * with a record, its position might advance to 6 (or higher) directly. Similarly, if the consumer's position is 5,
   * but there is no record with offset 5, the consumer will return the record with the next higher offset.
   * There are actually two notions of position relevant to the user of the consumer:
   * <p>
   * The {@link #position(TopicPartition) position} of the consumer gives the offset of the next record that will be given
   * out. It will be one larger than the highest offset the consumer has seen in that partition. It automatically advances
   * every time the consumer receives messages in a call to {@link #poll(Duration)}.
   * <p>
   * The {@link #commitSync() committed position} is the last offset that has been stored securely. Should the
   * process fail and restart, this is the offset that the consumer will recover to. The consumer can either automatically commit
   * offsets periodically; or it can choose to control this committed position manually by calling one of the commit APIs
   * (e.g. {@link #commitSync() commitSync} and {@link #commitAsync(OffsetCommitCallback) commitAsync}).
   * <p>
   * This distinction gives the consumer control over when a record is considered consumed. It is discussed in further
   * detail below.
   *
   * <h3><a name="consumergroups">Consumer Groups and Topic Subscriptions</a></h3>
   *
   * Kafka uses the concept of <i>consumer groups</i> to allow a pool of processes to divide the work of consuming and
   * processing records. These processes can either be running on the same machine or they can be
   * distributed over many machines to provide scalability and fault tolerance for processing. All consumer instances
   * sharing the same {@code group.id} will be part of the same consumer group.
   * <p>
   * Each consumer in a group can dynamically set the list of topics it wants to subscribe to through one of the
   * {@link #subscribe(Collection, ConsumerRebalanceListener) subscribe} APIs. Kafka will deliver each message in the
   * subscribed topics to one process in each consumer group. This is achieved by balancing the partitions between all
   * members in the consumer group so that each partition is assigned to exactly one consumer in the group. So if there
   * is a topic with four partitions, and a consumer group with two processes, each process would consume from two partitions.
   * <p>
   * Membership in a consumer group is maintained dynamically: if a process fails, the partitions assigned to it will
   * be reassigned to other consumers in the same group. Similarly, if a new consumer joins the group, partitions will be moved
   * from existing consumers to the new one. This is known as <i>rebalancing</i> the group and is discussed in more
   * detail <a href="#failuredetection">below</a>. Group rebalancing is also used when new partitions are added
   * to one of the subscribed topics or when a new topic matching a {@link #subscribe(Pattern, ConsumerRebalanceListener) subscribed regex}
   * is created. The group will automatically detect the new partitions through periodic metadata refreshes and
   * assign them to members of the group.
   * <p>
   * Conceptually you can think of a consumer group as being a single logical subscriber that happens to be made up of
   * multiple processes. As a multi-subscriber system, Kafka naturally supports having any number of consumer groups for a
   * given topic without duplicating data (additional consumers are actually quite cheap).
   * <p>
   * This is a slight generalization of the functionality that is common in messaging systems. To get semantics similar to
   * a queue in a traditional messaging system all processes would be part of a single consumer group and hence record
   * delivery would be balanced over the group like with a queue. Unlike a traditional messaging system, though, you can
   * have multiple such groups. To get semantics similar to pub-sub in a traditional messaging system each process would
   * have its own consumer group, so each process would subscribe to all the records published to the topic.
   * <p>
   * In addition, when group reassignment happens automatically, consumers can be notified through a {@link ConsumerRebalanceListener},
   * which allows them to finish necessary application-level logic such as state cleanup, manual offset
   * commits, etc. See <a href="#rebalancecallback">Storing Offsets Outside Kafka</a> for more details.
   * <p>
   * It is also possible for the consumer to <a href="#manualassignment">manually assign</a> specific partitions
   * (similar to the older "simple" consumer) using {@link #assign(Collection)}. In this case, dynamic partition
   * assignment and consumer group coordination will be disabled.
   *
   * <h3><a name="failuredetection">Detecting Consumer Failures</a></h3>
   *
   * After subscribing to a set of topics, the consumer will automatically join the group when {@link #poll(Duration)} is
   * invoked. The poll API is designed to ensure consumer liveness. As long as you continue to call poll, the consumer
   * will stay in the group and continue to receive messages from the partitions it was assigned. Underneath the covers,
   * the consumer sends periodic heartbeats to the server. If the consumer crashes or is unable to send heartbeats for
   * a duration of {@code session.timeout.ms}, then the consumer will be considered dead and its partitions will
   * be reassigned.
   * <p>
   * It is also possible that the consumer could encounter a "livelock" situation where it is continuing
   * to send heartbeats, but no progress is being made. To prevent the consumer from holding onto its partitions
   * indefinitely in this case, we provide a liveness detection mechanism using the {@code max.poll.interval.ms}
   * setting. Basically if you don't call poll at least as frequently as the configured max interval,
   * then the client will proactively leave the group so that another consumer can take over its partitions. When this happens,
   * you may see an offset commit failure (as indicated by a {@link CommitFailedException} thrown from a call to {@link #commitSync()}).
   * This is a safety mechanism which guarantees that only active members of the group are able to commit offsets.
   * So to stay in the group, you must continue to call poll.
   * <p>
   * The consumer provides two configuration settings to control the behavior of the poll loop:
   * <ol>
   *     <li><code>max.poll.interval.ms</code>: By increasing the interval between expected polls, you can give
   *     the consumer more time to handle a batch of records returned from {@link #poll(Duration)}. The drawback
   *     is that increasing this value may delay a group rebalance since the consumer will only join the rebalance
   *     inside the call to poll. You can use this setting to bound the time to finish a rebalance, but
   *     you risk slower progress if the consumer cannot actually call {@link #poll(Duration) poll} often enough.</li>
   *     <li><code>max.poll.records</code>: Use this setting to limit the total records returned from a single
   *     call to poll. This can make it easier to predict the maximum that must be handled within each poll
   *     interval. By tuning this value, you may be able to reduce the poll interval, which will reduce the
   *     impact of group rebalancing.</li>
   * </ol>
   * <p>
   * For use cases where message processing time varies unpredictably, neither of these options may be sufficient.
   * The recommended way to handle these cases is to move message processing to another thread, which allows
   * the consumer to continue calling {@link #poll(Duration) poll} while the processor is still working.
   * Some care must be taken to ensure that committed offsets do not get ahead of the actual position.
   * Typically, you must disable automatic commits and manually commit processed offsets for records only after the
   * thread has finished handling them (depending on the delivery semantics you need).
   * Note also that you will need to {@link #pause(Collection) pause} the partition so that no new records are received
   * from poll until after thread has finished handling those previously returned.
   *
   * <h3>Usage Examples</h3>
   * The consumer APIs offer flexibility to cover a variety of consumption use cases. Here are some examples to
   * demonstrate how to use them.
   *
   * <h4>Automatic Offset Committing</h4>
   * This example demonstrates a simple usage of Kafka's consumer api that relies on automatic offset committing.
   * <p>
   * <pre>
   *     Properties props = new Properties();
   *     props.setProperty(&quot;bootstrap.servers&quot;, &quot;localhost:9092&quot;);
   *     props.setProperty(&quot;group.id&quot;, &quot;test&quot;);
   *     props.setProperty(&quot;enable.auto.commit&quot;, &quot;true&quot;);
   *     props.setProperty(&quot;auto.commit.interval.ms&quot;, &quot;1000&quot;);
   *     props.setProperty(&quot;key.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;);
   *     props.setProperty(&quot;value.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;);
   *     KafkaConsumer&lt;String, String&gt; consumer = new KafkaConsumer&lt;&gt;(props);
   *     consumer.subscribe(Arrays.asList(&quot;foo&quot;, &quot;bar&quot;));
   *     while (true) {
   *         ConsumerRecords&lt;String, String&gt; records = consumer.poll(Duration.ofMillis(100));
   *         for (ConsumerRecord&lt;String, String&gt; record : records)
   *             System.out.printf(&quot;offset = %d, key = %s, value = %s%n&quot;, record.offset(), record.key(), record.value());
   *     }
   * </pre>
   *
   * The connection to the cluster is bootstrapped by specifying a list of one or more brokers to contact using the
   * configuration {@code bootstrap.servers}. This list is just used to discover the rest of the brokers in the
   * cluster and need not be an exhaustive list of servers in the cluster (though you may want to specify more than one in
   * case there are servers down when the client is connecting).
   * <p>
   * Setting {@code enable.auto.commit} means that offsets are committed automatically with a frequency controlled by
   * the config {@code auto.commit.interval.ms}.
   * <p>
   * In this example the consumer is subscribing to the topics <i>foo</i> and <i>bar</i> as part of a group of consumers
   * called <i>test</i> as configured with {@code group.id}.
   * <p>
   * The deserializer settings specify how to turn bytes into objects. For example, by specifying string deserializers, we
   * are saying that our record's key and value will just be simple strings.
   *
   * <h4>Manual Offset Control</h4>
   *
   * Instead of relying on the consumer to periodically commit consumed offsets, users can also control when records
   * should be considered as consumed and hence commit their offsets. This is useful when the consumption of the messages
   * is coupled with some processing logic and hence a message should not be considered as consumed until it is completed processing.
  
   * <p>
   * <pre>
   *     Properties props = new Properties();
   *     props.setProperty(&quot;bootstrap.servers&quot;, &quot;localhost:9092&quot;);
   *     props.setProperty(&quot;group.id&quot;, &quot;test&quot;);
   *     props.setProperty(&quot;enable.auto.commit&quot;, &quot;false&quot;);
   *     props.setProperty(&quot;key.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;);
   *     props.setProperty(&quot;value.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;);
   *     KafkaConsumer&lt;String, String&gt; consumer = new KafkaConsumer&lt;&gt;(props);
   *     consumer.subscribe(Arrays.asList(&quot;foo&quot;, &quot;bar&quot;));
   *     final int minBatchSize = 200;
   *     List&lt;ConsumerRecord&lt;String, String&gt;&gt; buffer = new ArrayList&lt;&gt;();
   *     while (true) {
   *         ConsumerRecords&lt;String, String&gt; records = consumer.poll(Duration.ofMillis(100));
   *         for (ConsumerRecord&lt;String, String&gt; record : records) {
   *             buffer.add(record);
   *         }
   *         if (buffer.size() &gt;= minBatchSize) {
   *             insertIntoDb(buffer);
   *             consumer.commitSync();
   *             buffer.clear();
   *         }
   *     }
   * </pre>
   *
   * In this example we will consume a batch of records and batch them up in memory. When we have enough records
   * batched, we will insert them into a database. If we allowed offsets to auto commit as in the previous example, records
   * would be considered consumed after they were returned to the user in {@link #poll(Duration) poll}. It would then be
   * possible
   * for our process to fail after batching the records, but before they had been inserted into the database.
   * <p>
   * To avoid this, we will manually commit the offsets only after the corresponding records have been inserted into the
   * database. This gives us exact control of when a record is considered consumed. This raises the opposite possibility:
   * the process could fail in the interval after the insert into the database but before the commit (even though this
   * would likely just be a few milliseconds, it is a possibility). In this case the process that took over consumption
   * would consume from last committed offset and would repeat the insert of the last batch of data. Used in this way
   * Kafka provides what is often called "at-least-once" delivery guarantees, as each record will likely be delivered one
   * time but in failure cases could be duplicated.
   * <p>
   * <b>Note: Using automatic offset commits can also give you "at-least-once" delivery, but the requirement is that
   * you must consume all data returned from each call to {@link #poll(Duration)} before any subsequent calls, or before
   * {@link #close() closing} the consumer. If you fail to do either of these, it is possible for the committed offset
   * to get ahead of the consumed position, which results in missing records. The advantage of using manual offset
   * control is that you have direct control over when a record is considered "consumed."</b>
   * <p>
   * The above example uses {@link #commitSync() commitSync} to mark all received records as committed. In some cases
   * you may wish to have even finer control over which records have been committed by specifying an offset explicitly.
   * In the example below we commit offset after we finish handling the records in each partition.
   * <p>
   * <pre>
   *     try {
   *         while(running) {
   *             ConsumerRecords&lt;String, String&gt; records = consumer.poll(Duration.ofMillis(Long.MAX_VALUE));
   *             for (TopicPartition partition : records.partitions()) {
   *                 List&lt;ConsumerRecord&lt;String, String&gt;&gt; partitionRecords = records.records(partition);
   *                 for (ConsumerRecord&lt;String, String&gt; record : partitionRecords) {
   *                     System.out.println(record.offset() + &quot;: &quot; + record.value());
   *                 }
   *                 consumer.commitSync(Collections.singletonMap(partition, records.nextOffsets().get(partition)));
   *             }
   *         }
   *     } finally {
   *       consumer.close();
   *     }
   * </pre>
   *
   * <b>Note: The committed offset should always be the offset of the next message that your application will read.</b>
   * Thus, when calling {@link #commitSync(Map) commitSync(offsets)} you should use {@code nextRecordToBeProcessed.offset()}
   * or if {@link ConsumerRecords} is exhausted already {@link ConsumerRecords#nextOffsets()} instead.
   * You should also add the leader epoch as commit metadata, which can be obtained from
   * {@link ConsumerRecord#leaderEpoch()} or {@link ConsumerRecords#nextOffsets()}.
   *
   * <h4><a name="manualassignment">Manual Partition Assignment</a></h4>
   *
   * In the previous examples, we subscribed to the topics we were interested in and let Kafka dynamically assign a
   * fair share of the partitions for those topics based on the active consumers in the group. However, in
   * some cases you may need finer control over the specific partitions that are assigned. For example:
   * <p>
   * <ul>
   * <li>If the process is maintaining some kind of local state associated with that partition (like a
   * local on-disk key-value store), then it should only get records for the partition it is maintaining on disk.
   * <li>If the process itself is highly available and will be restarted if it fails (perhaps using a
   * cluster management framework like YARN, Mesos, or AWS facilities, or as part of a stream processing framework). In
   * this case there is no need for Kafka to detect the failure and reassign the partition since the consuming process
   * will be restarted on another machine.
   * </ul>
   * <p>
   * To use this mode, instead of subscribing to the topic using {@link #subscribe(Collection) subscribe}, you just call
   * {@link #assign(Collection)} with the full list of partitions that you want to consume.
   *
   * <pre>
   *     String topic = &quot;foo&quot;;
   *     TopicPartition partition0 = new TopicPartition(topic, 0);
   *     TopicPartition partition1 = new TopicPartition(topic, 1);
   *     consumer.assign(Arrays.asList(partition0, partition1));
   * </pre>
   *
   * Once assigned, you can call {@link #poll(Duration) poll} in a loop, just as in the preceding examples to consume
   * records. The group that the consumer specifies is still used for committing offsets, but now the set of partitions
   * will only change with another call to {@link #assign(Collection) assign}. Manual partition assignment does
   * not use group coordination, so consumer failures will not cause assigned partitions to be rebalanced. Each consumer
   * acts independently even if it shares a groupId with another consumer. To avoid offset commit conflicts, you should
   * usually ensure that the groupId is unique for each consumer instance.
   * <p>
   * Note that it isn't possible to mix manual partition assignment (i.e. using {@link #assign(Collection) assign})
   * with dynamic partition assignment through topic subscription (i.e. using {@link #subscribe(Collection) subscribe}).
   *
   * <h4><a name="rebalancecallback">Storing Offsets Outside Kafka</h4>
   *
   * The consumer application need not use Kafka's built-in offset storage, it can store offsets in a store of its own
   * choosing. The primary use case for this is allowing the application to store both the offset and the results of the
   * consumption in the same system in a way that both the results and offsets are stored atomically. This is not always
   * possible, but when it is it will make the consumption fully atomic and give "exactly once" semantics that are
   * stronger than the default "at-least once" semantics you get with Kafka's offset commit functionality.
   * <p>
   * Here are a couple of examples of this type of usage:
   * <ul>
   * <li>If the results of the consumption are being stored in a relational database, storing the offset in the database
   * as well can allow committing both the results and offset in a single transaction. Thus either the transaction will
   * succeed and the offset will be updated based on what was consumed or the result will not be stored and the offset
   * won't be updated.
   * <li>If the results are being stored in a local store it may be possible to store the offset there as well. For
   * example a search index could be built by subscribing to a particular partition and storing both the offset and the
   * indexed data together. If this is done in a way that is atomic, it is often possible to have it be the case that even
   * if a crash occurs that causes unsync'd data to be lost, whatever is left has the corresponding offset stored as well.
   * This means that in this case the indexing process that comes back having lost recent updates just resumes indexing
   * from what it has ensuring that no updates are lost.
   * </ul>
   * <p>
   * Each record comes with its own offset, so to manage your own offset you just need to do the following:
   *
   * <ul>
   * <li>Configure <code>enable.auto.commit=false</code>
   * <li>Use the offset provided with each {@link ConsumerRecord} to save your position.
   * <li>On restart restore the position of the consumer using {@link #seek(TopicPartition, long)}.
   * </ul>
   *
   * <p>
   * This type of usage is simplest when the partition assignment is also done manually (this would be likely in the
   * search index use case described above). If the partition assignment is done automatically special care is
   * needed to handle the case where partition assignments change. This can be done by providing a
   * {@link ConsumerRebalanceListener} instance in the call to {@link #subscribe(Collection, ConsumerRebalanceListener)}
   * and {@link #subscribe(Pattern, ConsumerRebalanceListener)}.
   * For example, when partitions are taken from a consumer the consumer will want to commit its offset for those partitions by
   * implementing {@link ConsumerRebalanceListener#onPartitionsRevoked(Collection)}. When partitions are assigned to a
   * consumer, the consumer will want to look up the offset for those new partitions and correctly initialize the consumer
   * to that position by implementing {@link ConsumerRebalanceListener#onPartitionsAssigned(Collection)}.
   * <p>
   * Another common use for {@link ConsumerRebalanceListener} is to flush any caches the application maintains for
   * partitions that are moved elsewhere.
   *
   * <h4>Controlling The Consumer's Position</h4>
   *
   * In most use cases the consumer will simply consume records from beginning to end, periodically committing its
   * position (either automatically or manually). However Kafka allows the consumer to manually control its position,
   * moving forward or backwards in a partition at will. This means a consumer can re-consume older records, or skip to
   * the most recent records without actually consuming the intermediate records.
   * <p>
   * There are several instances where manually controlling the consumer's position can be useful.
   * <p>
   * One case is for time-sensitive record processing it may make sense for a consumer that falls far enough behind to not
   * attempt to catch up processing all records, but rather just skip to the most recent records.
   * <p>
   * Another use case is for a system that maintains local state as described in the previous section. In such a system
   * the consumer will want to initialize its position on start-up to whatever is contained in the local store. Likewise
   * if the local state is destroyed (say because the disk is lost) the state may be recreated on a new machine by
   * re-consuming all the data and recreating the state (assuming that Kafka is retaining sufficient history).
   * <p>
   * Kafka allows specifying the position using {@link #seek(TopicPartition, long)} to specify the new position. Special
   * methods for seeking to the earliest and latest offset the server maintains are also available (
   * {@link #seekToBeginning(Collection)} and {@link #seekToEnd(Collection)} respectively).
   *
   * <h4>Consumption Flow Control</h4>
   *
   * If a consumer is assigned multiple partitions to fetch data from, it will try to consume from all of them at the same time,
   * effectively giving these partitions the same priority for consumption. However in some cases consumers may want to
   * first focus on fetching from some subset of the assigned partitions at full speed, and only start fetching other partitions
   * when these partitions have few or no data to consume.
   *
   * <p>
   * One of such cases is stream processing, where processor fetches from two topics and performs the join on these two streams.
   * When one of the topics is long lagging behind the other, the processor would like to pause fetching from the ahead topic
   * in order to get the lagging stream to catch up. Another example is bootstrapping upon consumer starting up where there are
   * a lot of history data to catch up, the applications usually want to get the latest data on some of the topics before consider
   * fetching other topics.
   *
   * <p>
   * Kafka supports dynamic controlling of consumption flows by using {@link #pause(Collection)} and {@link #resume(Collection)}
   * to pause the consumption on the specified assigned partitions and resume the consumption
   * on the specified paused partitions respectively in the future {@link #poll(Duration)} calls.
   *
   * <h3>Reading Transactional Messages</h3>
   *
   * <p>
   * Transactions were introduced in Kafka 0.11.0 wherein applications can write to multiple topics and partitions atomically.
   * In order for this to work, consumers reading from these partitions should be configured to only read committed data.
   * This can be achieved by setting the {@code isolation.level=read_committed} in the consumer's configuration.
   *
   * <p>
   * In <code>read_committed</code> mode, the consumer will read only those transactional messages which have been
   * successfully committed. It will continue to read non-transactional messages as before. There is no client-side
   * buffering in <code>read_committed</code> mode. Instead, the end offset of a partition for a <code>read_committed</code>
   * consumer would be the offset of the first message in the partition belonging to an open transaction. This offset
   * is known as the 'Last Stable Offset'(LSO).</p>
   *
   * <p>
   * A {@code read_committed} consumer will only read up to the LSO and filter out any transactional
   * messages which have been aborted. The LSO also affects the behavior of {@link #seekToEnd(Collection)} and
   * {@link #endOffsets(Collection)} for {@code read_committed} consumers, details of which are in each method's documentation.
   * Finally, the fetch lag metrics are also adjusted to be relative to the LSO for {@code read_committed} consumers.
   *
   * <p>
   * Partitions with transactional messages will include commit or abort markers which indicate the result of a transaction.
   * There markers are not returned to applications, yet have an offset in the log. As a result, applications reading from
   * topics with transactional messages will see gaps in the consumed offsets. These missing messages would be the transaction
   * markers, and they are filtered out for consumers in both isolation levels. Additionally, applications using
   * {@code read_committed} consumers may also see gaps due to aborted transactions, since those messages would not
   * be returned by the consumer and yet would have valid offsets.
   *
   * <h3><a name="multithreaded">Multi-threaded Processing</a></h3>
   *
   * The Kafka consumer is NOT thread-safe. It is the responsibility of the user to ensure that multi-threaded access
   * is properly synchronized. Un-synchronized access will result in {@link ConcurrentModificationException}.
   *
   * <p>
   * The only exception to this rule is {@link #wakeup()}, which can safely be used from an external thread to
   * interrupt an active operation. In this case, a {@link org.apache.kafka.common.errors.WakeupException} will be
   * thrown from the thread blocking on the operation. This can be used to shutdown the consumer from another thread.
   * The following snippet shows the typical pattern:
   *
   * <pre>
   * public class KafkaConsumerRunner implements Runnable {
   *     private final AtomicBoolean closed = new AtomicBoolean(false);
   *     private final KafkaConsumer consumer;
   *
   *     public KafkaConsumerRunner(KafkaConsumer consumer) {
   *       this.consumer = consumer;
   *     }
   *
   *     {@literal}@Override
   *     public void run() {
   *         try {
   *             consumer.subscribe(Arrays.asList("topic"));
   *             while (!closed.get()) {
   *                 ConsumerRecords records = consumer.poll(Duration.ofMillis(10000));
   *                 // Handle new records
   *             }
   *         } catch (WakeupException e) {
   *             // Ignore exception if closing
   *             if (!closed.get()) throw e;
   *         } finally {
   *             consumer.close();
   *         }
   *     }
   *
   *     // Shutdown hook which can be called from a separate thread
   *     public void shutdown() {
   *         closed.set(true);
   *         consumer.wakeup();
   *     }
   * }
   * </pre>
   *
   * Then in a separate thread, the consumer can be shutdown by setting the closed flag and waking up the consumer.
   *
   * <p>
   * <pre>
   *     closed.set(true);
   *     consumer.wakeup();
   * </pre>
   *
   * <p>
   * Note that while it is possible to use thread interrupts instead of {@link #wakeup()} to abort a blocking operation
   * (in which case, {@link InterruptException} will be raised), we discourage their use since they may cause a clean
   * shutdown of the consumer to be aborted. Interrupts are mainly supported for those cases where using {@link #wakeup()}
   * is impossible, e.g. when a consumer thread is managed by code that is unaware of the Kafka client.
   *
   * <p>
   * We have intentionally avoided implementing a particular threading model for processing. This leaves several
   * options for implementing multi-threaded processing of records.
   *
   * <h4>1. One Consumer Per Thread</h4>
   *
   * A simple option is to give each thread its own consumer instance. Here are the pros and cons of this approach:
   * <ul>
   * <li><b>PRO</b>: It is the easiest to implement
   * <li><b>PRO</b>: It is often the fastest as no inter-thread co-ordination is needed
   * <li><b>PRO</b>: It makes in-order processing on a per-partition basis very easy to implement (each thread just
   * processes messages in the order it receives them).
   * <li><b>CON</b>: More consumers means more TCP connections to the cluster (one per thread). In general Kafka handles
   * connections very efficiently so this is generally a small cost.
   * <li><b>CON</b>: Multiple consumers means more requests being sent to the server and slightly less batching of data
   * which can cause some drop in I/O throughput.
   * <li><b>CON</b>: The number of total threads across all processes will be limited by the total number of partitions.
   * </ul>
   *
   * <h4>2. Decouple Consumption and Processing</h4>
   *
   * Another alternative is to have one or more consumer threads that do all data consumption and hands off
   * {@link ConsumerRecords} instances to a blocking queue consumed by a pool of processor threads that actually handle
   * the record processing.
   *
   * This option likewise has pros and cons:
   * <ul>
   * <li><b>PRO</b>: This option allows independently scaling the number of consumers and processors. This makes it
   * possible to have a single consumer that feeds many processor threads, avoiding any limitation on partitions.
   * <li><b>CON</b>: Guaranteeing order across the processors requires particular care as the threads will execute
   * independently an earlier chunk of data may actually be processed after a later chunk of data just due to the luck of
   * thread execution timing. For processing that has no ordering requirements this is not a problem.
   * <li><b>CON</b>: Manually committing the position becomes harder as it requires that all threads co-ordinate to ensure
   * that processing is complete for that partition.
   * </ul>
   *
   * There are many possible variations on this approach. For example each processor thread can have its own queue, and
   * the consumer threads can hash into these queues using the TopicPartition to ensure in-order consumption and simplify
   * commit.
   */
  public class KafkaConsumer<K, V> implements Consumer<K, V> {
  
      private static final ConsumerDelegateCreator CREATOR = new ConsumerDelegateCreator();
  
      private final ConsumerDelegate<K, V> delegate;
  
      /**
       * A consumer is instantiated by providing a set of key-value pairs as configuration. Valid configuration strings
       * are documented <a href="http://kafka.apache.org/documentation.html#consumerconfigs" >here</a>. Values can be
       * either strings or objects of the appropriate type (for example a numeric configuration would accept either the
       * string "42" or the integer 42).
       * <p>
       * Valid configuration strings are documented at {@link ConsumerConfig}.
       * <p>
       * Note: after creating a {@code KafkaConsumer} you must always {@link #close()} it to avoid resource leaks.
       *
       * @param configs The consumer configs
       */
      public KafkaConsumer(Map<String, Object> configs) {
          this(configs, null, null);
      }
  
      /**
       * A consumer is instantiated by providing a {@link java.util.Properties} object as configuration.
       * <p>
       * Valid configuration strings are documented at {@link ConsumerConfig}.
       * <p>
       * Note: after creating a {@code KafkaConsumer} you must always {@link #close()} it to avoid resource leaks.
       *
       * @param properties The consumer configuration properties
       */
      public KafkaConsumer(Properties properties) {
          this(properties, null, null);
      }
  
      /**
       * A consumer is instantiated by providing a {@link java.util.Properties} object as configuration, and a
       * key and a value {@link Deserializer}.
       * <p>
       * Valid configuration strings are documented at {@link ConsumerConfig}.
       * <p>
       * Note: after creating a {@code KafkaConsumer} you must always {@link #close()} it to avoid resource leaks.
       *
       * @param properties The consumer configuration properties
       * @param keyDeserializer The deserializer for key that implements {@link Deserializer}. The configure() method
       *            won't be called in the consumer when the deserializer is passed in directly.
       * @param valueDeserializer The deserializer for value that implements {@link Deserializer}. The configure() method
       *            won't be called in the consumer when the deserializer is passed in directly.
       */
      public KafkaConsumer(Properties properties,
                           Deserializer<K> keyDeserializer,
                           Deserializer<V> valueDeserializer) {
          this(propsToMap(properties), keyDeserializer, valueDeserializer);
      }
  
      /**
       * A consumer is instantiated by providing a set of key-value pairs as configuration, and a key and a value {@link Deserializer}.
       * <p>
       * Valid configuration strings are documented at {@link ConsumerConfig}.
       * <p>
       * Note: after creating a {@code KafkaConsumer} you must always {@link #close()} it to avoid resource leaks.
       *
       * @param configs The consumer configs
       * @param keyDeserializer The deserializer for key that implements {@link Deserializer}. The configure() method
       *            won't be called in the consumer when the deserializer is passed in directly.
       * @param valueDeserializer The deserializer for value that implements {@link Deserializer}. The configure() method
       *            won't be called in the consumer when the deserializer is passed in directly.
       */
      public KafkaConsumer(Map<String, Object> configs,
                           Deserializer<K> keyDeserializer,
                           Deserializer<V> valueDeserializer) {
          this(new ConsumerConfig(ConsumerConfig.appendDeserializerToConfig(configs, keyDeserializer, valueDeserializer)),
                  keyDeserializer, valueDeserializer);
      }
  
      KafkaConsumer(ConsumerConfig config, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) {
          delegate = CREATOR.create(config, keyDeserializer, valueDeserializer);
      }
  
      KafkaConsumer(LogContext logContext,
                    Time time,
                    ConsumerConfig config,
                    Deserializer<K> keyDeserializer,
                    Deserializer<V> valueDeserializer,
                    KafkaClient client,
                    SubscriptionState subscriptions,
                    ConsumerMetadata metadata,
                    List<ConsumerPartitionAssignor> assignors) {
          delegate = CREATOR.create(
              logContext,
              time,
              config,
              keyDeserializer,
              valueDeserializer,
              client,
              subscriptions,
              metadata,
              assignors
          );
      }
  
      /**
       * Get the set of partitions currently assigned to this consumer. If subscription happened by directly assigning
       * partitions using {@link #assign(Collection)} then this will simply return the same partitions that
       * were assigned. If topic subscription was used, then this will give the set of topic partitions currently assigned
       * to the consumer (which may be none if the assignment hasn't happened yet, or the partitions are in the
       * process of getting reassigned).
       * @return The set of partitions currently assigned to this consumer
       */
      public Set<TopicPartition> assignment() {
          return delegate.assignment();
      }
  
      /**
       * Get the current subscription. Will return the same topics used in the most recent call to
       * {@link #subscribe(Collection, ConsumerRebalanceListener)}, or an empty set if no such call has been made.
       * @return The set of topics currently subscribed to
       */
      public Set<String> subscription() {
          return delegate.subscription();
      }
  
      /**
       * Subscribe to the given list of topics to get dynamically
       * assigned partitions. <b>Topic subscriptions are not incremental. This list will replace the current
       * assignment (if there is one).</b> Note that it is not possible to combine topic subscription with group management
       * with manual partition assignment through {@link #assign(Collection)}.
       *
       * If the given list of topics is empty, it is treated the same as {@link #unsubscribe()}.
       *
       * <p>
       * As part of group management, the consumer will keep track of the list of consumers that belong to a particular
       * group and will trigger a rebalance operation if any one of the following events are triggered:
       * <ul>
       * <li>Number of partitions change for any of the subscribed topics
       * <li>A subscribed topic is created or deleted
       * <li>An existing member of the consumer group is shutdown or fails
       * <li>A new member is added to the consumer group
       * </ul>
       * <p>
       * When any of these events are triggered, the provided listener will be invoked first to indicate that
       * the consumer's assignment has been revoked, and then again when the new assignment has been received.
       * Note that rebalances will only occur during an active call to {@link #poll(Duration)}, so callbacks will
       * also only be invoked during that time.
       *
       * The provided listener will immediately override any listener set in a previous call to subscribe.
       * It is guaranteed, however, that the partitions revoked/assigned through this interface are from topics
       * subscribed in this call. See {@link ConsumerRebalanceListener} for more details.
       *
       * @param topics The list of topics to subscribe to
       * @param listener Non-null listener instance to get notifications on partition assignment/revocation for the
       *                 subscribed topics
       * @throws IllegalArgumentException If topics is null or contains null or empty elements, or if listener is null
       * @throws IllegalStateException If {@code subscribe()} is called previously with pattern, or assign is called
       *                               previously (without a subsequent call to {@link #unsubscribe()}), or if not
       *                               configured at-least one partition assignment strategy
       */
      @Override
      public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener) {
          delegate.subscribe(topics, listener);
      }
  
      /**
       * Subscribe to the given list of topics to get dynamically assigned partitions.
       * <b>Topic subscriptions are not incremental. This list will replace the current
       * assignment (if there is one).</b> It is not possible to combine topic subscription with group management
       * with manual partition assignment through {@link #assign(Collection)}.
       *
       * If the given list of topics is empty, it is treated the same as {@link #unsubscribe()}.
       *
       * <p>
       * This is a short-hand for {@link #subscribe(Collection, ConsumerRebalanceListener)}, which
       * uses a no-op listener. If you need the ability to seek to particular offsets, you should prefer
       * {@link #subscribe(Collection, ConsumerRebalanceListener)}, since group rebalances will cause partition offsets
       * to be reset. You should also provide your own listener if you are doing your own offset
       * management since the listener gives you an opportunity to commit offsets before a rebalance finishes.
       *
       * @param topics The list of topics to subscribe to
       * @throws IllegalArgumentException If topics is null or contains null or empty elements
       * @throws IllegalStateException If {@code subscribe()} is called previously with pattern, or assign is called
       *                               previously (without a subsequent call to {@link #unsubscribe()}), or if not
       *                               configured at-least one partition assignment strategy
       */
      @Override
      public void subscribe(Collection<String> topics) {
          delegate.subscribe(topics);
      }
  
      /**
       * Subscribe to all topics matching specified pattern to get dynamically assigned partitions.
       * The pattern matching will be done periodically against all topics existing at the time of check.
       * This can be controlled through the {@code metadata.max.age.ms} configuration: by lowering
       * the max metadata age, the consumer will refresh metadata more often and check for matching topics.
       * <p>
       * See {@link #subscribe(Collection, ConsumerRebalanceListener)} for details on the
       * use of the {@link ConsumerRebalanceListener}. Generally rebalances are triggered when there
       * is a change to the topics matching the provided pattern and when consumer group membership changes.
       * Group rebalances only take place during an active call to {@link #poll(Duration)}.
       *
       * @param pattern Pattern to subscribe to
       * @param listener Non-null listener instance to get notifications on partition assignment/revocation for the
       *                 subscribed topics
       * @throws IllegalArgumentException If pattern or listener is null
       * @throws IllegalStateException If {@code subscribe()} is called previously with topics, or assign is called
       *                               previously (without a subsequent call to {@link #unsubscribe()}), or if not
       *                               configured at-least one partition assignment strategy
       */
      @Override
      public void subscribe(Pattern pattern, ConsumerRebalanceListener listener) {
          delegate.subscribe(pattern, listener);
      }
  
      /**
       * Subscribe to all topics matching specified pattern to get dynamically assigned partitions.
       * The pattern matching will be done periodically against topics existing at the time of check.
       * <p>
       * This is a short-hand for {@link #subscribe(Pattern, ConsumerRebalanceListener)}, which
       * uses a no-op listener. If you need the ability to seek to particular offsets, you should prefer
       * {@link #subscribe(Pattern, ConsumerRebalanceListener)}, since group rebalances will cause partition offsets
       * to be reset. You should also provide your own listener if you are doing your own offset
       * management since the listener gives you an opportunity to commit offsets before a rebalance finishes.
       *
       * @param pattern Pattern to subscribe to
       * @throws IllegalArgumentException If pattern is null
       * @throws IllegalStateException If {@code subscribe()} is called previously with topics, or assign is called
       *                               previously (without a subsequent call to {@link #unsubscribe()}), or if not
       *                               configured at-least one partition assignment strategy
       */
      @Override
      public void subscribe(Pattern pattern) {
          delegate.subscribe(pattern);
      }
  
      /**
       * Subscribe to all topics matching the specified pattern, to get dynamically assigned partitions.
       * The pattern matching will be done periodically against all topics. This is only supported under the
       * CONSUMER group protocol (see {@link ConsumerConfig#GROUP_PROTOCOL_CONFIG}).
       * <p>
       * If the provided pattern is not compatible with Google RE2/J, an {@link InvalidRegularExpression} will be
       * eventually thrown on a call to {@link #poll(Duration)} following this call to subscribe.
       * <p>
       * See {@link #subscribe(Collection, ConsumerRebalanceListener)} for details on the
       * use of the {@link ConsumerRebalanceListener}. Generally, rebalances are triggered when there
       * is a change to the topics matching the provided pattern and when consumer group membership changes.
       * Group rebalances only take place during an active call to {@link #poll(Duration)}.
       *
       * @param pattern  Pattern to subscribe to, that must be compatible with Google RE2/J.
       * @param listener Non-null listener instance to get notifications on partition assignment/revocation for the
       *                 subscribed topics.
       * @throws IllegalArgumentException If pattern is null or empty, or if the listener is null.
       * @throws IllegalStateException    If {@code subscribe()} is called previously with topics, or assign is called
       *                                  previously (without a subsequent call to {@link #unsubscribe()}).
       */
      @Override
      public void subscribe(SubscriptionPattern pattern, ConsumerRebalanceListener listener) {
          delegate.subscribe(pattern, listener);
      }
  
      /**
       * Subscribe to all topics matching the specified pattern, to get dynamically assigned partitions.
       * The pattern matching will be done periodically against topics. This is only supported under the
       * CONSUMER group protocol (see {@link ConsumerConfig#GROUP_PROTOCOL_CONFIG})
       * <p>
       * If the provided pattern is not compatible with Google RE2/J, an {@link InvalidRegularExpression} will be
       * eventually thrown on a call to {@link #poll(Duration)} following this call to subscribe.
       * <p>
       * This is a short-hand for {@link #subscribe(Pattern, ConsumerRebalanceListener)}, which
       * uses a no-op listener. If you need the ability to seek to particular offsets, you should prefer
       * {@link #subscribe(Pattern, ConsumerRebalanceListener)}, since group rebalances will cause partition offsets
       * to be reset. You should also provide your own listener if you are doing your own offset
       * management since the listener gives you an opportunity to commit offsets before a rebalance finishes.
       *
       * @param pattern Pattern to subscribe to, that must be compatible with Google RE2/J.
       * @throws IllegalArgumentException If pattern is null or empty.
       * @throws IllegalStateException    If {@code subscribe()} is called previously with topics, or assign is called
       *                                  previously (without a subsequent call to {@link #unsubscribe()}).
       */
      @Override
      public void subscribe(SubscriptionPattern pattern) {
          delegate.subscribe(pattern);
      }
  
      /**
       * Unsubscribe from topics currently subscribed with {@link #subscribe(Collection)} or {@link #subscribe(Pattern)}.
       * This also clears any partitions directly assigned through {@link #assign(Collection)}.
       *
       * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors (e.g. rebalance callback errors)
       */
      public void unsubscribe() {
          delegate.unsubscribe();
      }
  
      /**
       * Manually assign a list of partitions to this consumer. This interface does not allow for incremental assignment
       * and will replace the previous assignment (if there is one).
       * <p>
       * If the given list of topic partitions is empty, it is treated the same as {@link #unsubscribe()}.
       * <p>
       * Manual topic assignment through this method does not use the consumer's group management
       * functionality. As such, there will be no rebalance operation triggered when group membership or cluster and topic
       * metadata change. Note that it is not possible to use both manual partition assignment with {@link #assign(Collection)}
       * and group assignment with {@link #subscribe(Collection, ConsumerRebalanceListener)}.
       * <p>
       * If auto-commit is enabled, an async commit (based on the old assignment) will be triggered before the new
       * assignment replaces the old one.
       *
       * @param partitions The list of partitions to assign this consumer
       * @throws IllegalArgumentException If partitions is null or contains null or empty topics
       * @throws IllegalStateException If {@code subscribe()} is called previously with topics or pattern
       *                               (without a subsequent call to {@link #unsubscribe()})
       */
      @Override
      public void assign(Collection<TopicPartition> partitions) {
          delegate.assign(partitions);
      }
  
      /**
       * Fetch data for the topics or partitions specified using one of the subscribe/assign APIs. It is an error to not have
       * subscribed to any topics or partitions before polling for data.
       * <p>
       * On each poll, consumer will try to use the last consumed offset as the starting offset and fetch sequentially. The last
       * consumed offset can be manually set through {@link #seek(TopicPartition, long)} or automatically set as the last committed
       * offset for the subscribed list of partitions
       *
       * <p>
       * This method returns immediately if there are records available or if the position advances past control records
       * or aborted transactions when isolation.level=read_committed.
       * Otherwise, it will await the passed timeout. If the timeout expires, an empty record set will be returned.
       * Note that this method may block beyond the timeout in order to execute custom
       * {@link ConsumerRebalanceListener} callbacks.
       *
       *
       * @param timeout The maximum time to block (must not be greater than {@link Long#MAX_VALUE} milliseconds)
       *
       * @return map of topic to records since the last fetch for the subscribed list of topics and partitions
       *
       * @throws org.apache.kafka.clients.consumer.InvalidOffsetException if the offset for a partition or set of
       *             partitions is undefined or out of range and no offset reset policy has been configured
       * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
       *             function is called
       * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
       *             this function is called
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if caller lacks Read access to any of the subscribed
       *             topics or to the configured groupId. See the exception for more details
       * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors (e.g. invalid groupId or
       *             session timeout, errors deserializing key/value pairs, your rebalance callback thrown exceptions,
       *             or any new error cases in future versions)
       * @throws java.lang.IllegalArgumentException if the timeout value is negative
       * @throws java.lang.IllegalStateException if the consumer is not subscribed to any topics or manually assigned any
       *             partitions to consume from
       * @throws java.lang.ArithmeticException if the timeout is greater than {@link Long#MAX_VALUE} milliseconds.
       * @throws org.apache.kafka.common.errors.InvalidTopicException if the current subscription contains any invalid
       *             topic (per {@link org.apache.kafka.common.internals.Topic#validate(String)})
       * @throws org.apache.kafka.common.errors.UnsupportedVersionException if the consumer attempts to fetch stable offsets
       *             when the broker doesn't support this feature. Also, if the consumer attempts to subscribe to a
       *             SubscriptionPattern via {@link #subscribe(SubscriptionPattern)} or
       *             {@link #subscribe(SubscriptionPattern, ConsumerRebalanceListener)} and the broker doesn't
       *             support this feature.
       * @throws org.apache.kafka.common.errors.FencedInstanceIdException if this consumer instance gets fenced by broker.
       */
      @Override
      public ConsumerRecords<K, V> poll(final Duration timeout) {
          return delegate.poll(timeout);
      }
  
      /**
       * Commit offsets returned on the last {@link #poll(Duration) poll()} for all the subscribed list of topics and
       * partitions.
       * <p>
       * This commits offsets only to Kafka. The offsets committed using this API will be used on the first fetch after
       * every rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API
       * should not be used.
       * <p>
       * This is a synchronous commit and will block until either the commit succeeds, an unrecoverable error is
       * encountered (in which case it is thrown to the caller), or the timeout specified by {@code default.api.timeout.ms} expires
       * (in which case a {@link org.apache.kafka.common.errors.TimeoutException} is thrown to the caller).
       * <p>
       * Note that asynchronous offset commits sent previously with the {@link #commitAsync(OffsetCommitCallback)}
       * (or similar) are guaranteed to have their callbacks invoked prior to completion of this method,
       * but only when the consumer is using the consumer group protocol.
       *
       * @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried.
       *             This fatal error can only occur if you are using automatic group management with {@link #subscribe(Collection)},
       *             or if there is an active group with the same <code>group.id</code> which is using group management. In such cases,
       *             when you are trying to commit to partitions that are no longer assigned to this consumer because the
       *             consumer is for example no longer part of the group this exception would be thrown.
       * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the consumer instance is in the middle of a rebalance
       *            so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first
       *            complete the rebalance by calling {@link #poll(Duration)} and commit can be reconsidered afterwards.
       *            NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed,
       *            and also for those partitions that are still assigned their fetch positions may have changed too
       *            if more records are returned from the {@link #poll(Duration)} call.
       * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
       *             function is called
       * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
       *             this function is called
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the
       *             configured groupId. See the exception for more details
       * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors (e.g. if offset metadata
       *             is too large or if the topic does not exist).
       * @throws org.apache.kafka.common.errors.TimeoutException if the timeout specified by {@code default.api.timeout.ms} expires
       *            before successful completion of the offset commit
       * @throws org.apache.kafka.common.errors.FencedInstanceIdException if this consumer is using the classic group protocol
       *            and this instance gets fenced by broker.
       */
      @Override
      public void commitSync() {
          delegate.commitSync();
      }
  
      /**
       * Commit offsets returned on the last {@link #poll(Duration) poll()} for all the subscribed list of topics and
       * partitions.
       * <p>
       * This commits offsets only to Kafka. The offsets committed using this API will be used on the first fetch after
       * every rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API
       * should not be used.
       * <p>
       * This is a synchronous commit and will block until either the commit succeeds, an unrecoverable error is
       * encountered (in which case it is thrown to the caller), or the passed timeout expires.
       * <p>
       * Note that asynchronous offset commits sent previously with the {@link #commitAsync(OffsetCommitCallback)}
       * (or similar) are guaranteed to have their callbacks invoked prior to completion of this method,
       * but only when the consumer is using the consumer group protocol.
       *
       * @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried.
       *             This can only occur if you are using automatic group management with {@link #subscribe(Collection)},
       *             or if there is an active group with the same <code>group.id</code> which is using group management. In such cases,
       *             when you are trying to commit to partitions that are no longer assigned to this consumer because the
       *             consumer is for example no longer part of the group this exception would be thrown.
       * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the consumer instance is in the middle of a rebalance
       *            so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first
       *            complete the rebalance by calling {@link #poll(Duration)} and commit can be reconsidered afterwards.
       *            NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed,
       *            and also for those partitions that are still assigned their fetch positions may have changed too
       *            if more records are returned from the {@link #poll(Duration)} call.
       * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
       *             function is called
       * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
       *             this function is called
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the
       *             configured groupId. See the exception for more details
       * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors (e.g. if offset metadata
       *             is too large or if the topic does not exist).
       * @throws org.apache.kafka.common.errors.TimeoutException if the timeout expires before successful completion
       *            of the offset commit
       * @throws org.apache.kafka.common.errors.FencedInstanceIdException if this consumer is using the classic group protocol
       *            and this instance gets fenced by broker.
       */
      @Override
      public void commitSync(Duration timeout) {
          delegate.commitSync(timeout);
      }
  
      /**
       * Commit the specified offsets for the specified list of topics and partitions.
       * <p>
       * This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every
       * rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API
       * should not be used. The committed offset should be the next message your application will consume,
       * i.e. {@code nextRecordToBeProcessed.offset()} (or {@link ConsumerRecords#nextOffsets()}).
       * You should also add the leader epoch as commit metadata, which can be obtained from
       * {@link ConsumerRecord#leaderEpoch()} or {@link ConsumerRecords#nextOffsets()}.
       * If automatic group management with {@link #subscribe(Collection)} is used,
       * then the committed offsets must belong to the currently auto-assigned partitions.
       * <p>
       * This is a synchronous commit and will block until either the commit succeeds or an unrecoverable error is
       * encountered (in which case it is thrown to the caller), or the timeout specified by {@code default.api.timeout.ms} expires
       * (in which case a {@link org.apache.kafka.common.errors.TimeoutException} is thrown to the caller).
       * <p>
       * Note that asynchronous offset commits sent previously with the {@link #commitAsync(OffsetCommitCallback)}
       * (or similar) are guaranteed to have their callbacks invoked prior to completion of this method,
       * but only when the consumer is using the consumer group protocol.
       *
       * @param offsets A map of offsets by partition with associated metadata. This map will be copied internally, so it
       *                is safe to mutate the map after returning.
       * @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried.
       *             This can only occur if you are using automatic group management with {@link #subscribe(Collection)},
       *             or if there is an active group with the same <code>group.id</code> which is using group management. In such cases,
       *             when you are trying to commit to partitions that are no longer assigned to this consumer because the
       *             consumer is for example no longer part of the group this exception would be thrown.
       * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the consumer instance is in the middle of a rebalance
       *            so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first
       *            complete the rebalance by calling {@link #poll(Duration)} and commit can be reconsidered afterwards.
       *            NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed,
       *            and also for those partitions that are still assigned their fetch positions may have changed too
       *            if more records are returned from the {@link #poll(Duration)} call, so when you retry committing
       *            you should consider updating the passed in {@code offset} parameter.
       * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
       *             function is called
       * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
       *             this function is called
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the
       *             configured groupId. See the exception for more details
       * @throws java.lang.IllegalArgumentException if the committed offset is negative
       * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors (e.g. if offset metadata
       *             is too large or if the topic does not exist).
       * @throws org.apache.kafka.common.errors.TimeoutException if the timeout expires before successful completion
       *            of the offset commit
       * @throws org.apache.kafka.common.errors.FencedInstanceIdException if this consumer is using the classic group protocol
       *            and this instance gets fenced by broker.
       */
      @Override
      public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets) {
          delegate.commitSync(offsets);
      }
  
      /**
       * Commit the specified offsets for the specified list of topics and partitions.
       * <p>
       * This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every
       * rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API
       * should not be used. The committed offset should be the next message your application will consume,
       * i.e. {@code nextRecordToBeProcessed.offset()} (or {@link ConsumerRecords#nextOffsets()}).
       * You should also add the leader epoch as commit metadata, which can be obtained from
       * {@link ConsumerRecord#leaderEpoch()} or {@link ConsumerRecords#nextOffsets()}.
       * If automatic group management with {@link #subscribe(Collection)} is used,
       * then the committed offsets must belong to the currently auto-assigned partitions.
       * <p>
       * This is a synchronous commit and will block until either the commit succeeds, an unrecoverable error is
       * encountered (in which case it is thrown to the caller), or the timeout expires.
       * <p>
       * Note that asynchronous offset commits sent previously with the {@link #commitAsync(OffsetCommitCallback)}
       * (or similar) are guaranteed to have their callbacks invoked prior to completion of this method,
       * but only when the consumer is using the consumer group protocol.
       *
       * @param offsets A map of offsets by partition with associated metadata. This map will be copied internally, so it
       *                is safe to mutate the map after returning.
       * @param timeout The maximum amount of time to await completion of the offset commit
       * @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried.
       *             This can only occur if you are using automatic group management with {@link #subscribe(Collection)},
       *             or if there is an active group with the same <code>group.id</code> which is using group management. In such cases,
       *             when you are trying to commit to partitions that are no longer assigned to this consumer because the
       *             consumer is for example no longer part of the group this exception would be thrown.
       * @throws org.apache.kafka.common.errors.RebalanceInProgressException if the consumer instance is in the middle of a rebalance
       *            so it is not yet determined which partitions would be assigned to the consumer. In such cases you can first
       *            complete the rebalance by calling {@link #poll(Duration)} and commit can be reconsidered afterwards.
       *            NOTE when you reconsider committing after the rebalance, the assigned partitions may have changed,
       *            and also for those partitions that are still assigned their fetch positions may have changed too
       *            if more records are returned from the {@link #poll(Duration)} call, so when you retry committing
       *            you should consider updating the passed in {@code offset} parameter.
       * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
       *             function is called
       * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
       *             this function is called
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the
       *             configured groupId. See the exception for more details
       * @throws java.lang.IllegalArgumentException if the committed offset is negative
       * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors (e.g. if offset metadata
       *             is too large or if the topic does not exist).
       * @throws org.apache.kafka.common.errors.TimeoutException if the timeout expires before successful completion
       *            of the offset commit
       * @throws org.apache.kafka.common.errors.FencedInstanceIdException if this consumer is using the classic group protocol
       *            and this instance gets fenced by broker.
       */
      @Override
      public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets, final Duration timeout) {
          delegate.commitSync(offsets, timeout);
      }
  
      /**
       * Commit offsets returned on the last {@link #poll(Duration)} for all the subscribed list of topics and partition.
       * Same as {@link #commitAsync(OffsetCommitCallback) commitAsync(null)}
       * @throws org.apache.kafka.common.errors.FencedInstanceIdException if this consumer is using the classic group protocol
       *            and this instance gets fenced by broker.
       */
      @Override
      public void commitAsync() {
          delegate.commitAsync();
      }
  
      /**
       * Commit offsets returned on the last {@link #poll(Duration) poll()} for the subscribed list of topics and partitions.
       * <p>
       * This commits offsets only to Kafka. The offsets committed using this API will be used on the first fetch after
       * every rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API
       * should not be used.
       * <p>
       * This is an asynchronous call and will not block. Any errors encountered are either passed to the callback
       * (if provided) or discarded.
       * <p>
       * Offsets committed through multiple calls to this API are guaranteed to be sent in the same order as
       * the invocations. Corresponding commit callbacks are also invoked in the same order. Additionally note that
       * offsets committed through this API are guaranteed to complete before a subsequent call to {@link #commitSync()}
       * (and variants) returns.
       *
       * @param callback Callback to invoke when the commit completes
       * @throws org.apache.kafka.common.errors.FencedInstanceIdException if this consumer is using the classic group protocol
       *             and this instance gets fenced by broker.
       */
      @Override
      public void commitAsync(OffsetCommitCallback callback) {
          delegate.commitAsync(callback);
      }
  
      /**
       * Commit the specified offsets for the specified list of topics and partitions to Kafka.
       * <p>
       * This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every
       * rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API
       * should not be used. The committed offset should be the next message your application will consume,
       * i.e. {@code nextRecordToBeProcessed.offset()} (or {@link ConsumerRecords#nextOffsets()}).
       * You should also add the leader epoch as commit metadata, which can be obtained from
       * {@link ConsumerRecord#leaderEpoch()} or {@link ConsumerRecords#nextOffsets()}.
       * If automatic group management with {@link #subscribe(Collection)} is used,
       * then the committed offsets must belong to the currently auto-assigned partitions.
       * <p>
       * This is an asynchronous call and will not block. Any errors encountered are either passed to the callback
       * (if provided) or discarded.
       * <p>
       * Offsets committed through multiple calls to this API are guaranteed to be sent in the same order as
       * the invocations. Corresponding commit callbacks are also invoked in the same order. Additionally note that
       * offsets committed through this API are guaranteed to complete before a subsequent call to {@link #commitSync()}
       * (and variants) returns.
       *
       * @param offsets A map of offsets by partition with associated metadata. This map will be copied internally, so it
       *                is safe to mutate the map after returning.
       * @param callback Callback to invoke when the commit completes
       * @throws org.apache.kafka.common.errors.FencedInstanceIdException if this consumer is using the classic group protocol
       *             and this instance gets fenced by broker.
       */
      @Override
      public void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback) {
          delegate.commitAsync(offsets, callback);
      }
  
      /**
       * Overrides the fetch offsets that the consumer will use on the next {@link #poll(Duration) poll(timeout)}. If this API
       * is invoked for the same partition more than once, the latest offset will be used on the next poll(). Note that
       * you may lose data if this API is arbitrarily used in the middle of consumption, to reset the fetch offsets
       * <p>
       * The next Consumer Record which will be retrieved when poll() is invoked will have the offset specified, given that
       * a record with that offset exists (i.e. it is a valid offset).
       * <p>
       * {@link #seekToBeginning(Collection)} will go to the first offset in the topic.
       * seek(0) is equivalent to seekToBeginning for a TopicPartition with beginning offset 0,
       * assuming that there is a record at offset 0 still available.
       * {@link #seekToEnd(Collection)} is equivalent to seeking to the last offset of the partition, but behavior depends on
       * {@code isolation.level}, so see {@link #seekToEnd(Collection)} documentation for more details.
       * <p>
       * Seeking to the offset smaller than the log start offset or larger than the log end offset
       * means an invalid offset is reached.
       * Invalid offset behaviour is controlled by the {@code auto.offset.reset} property.
       * If this is set to "earliest", the next poll will return records from the starting offset.
       * If it is set to "latest", it will seek to the last offset (similar to seekToEnd()).
       * If it is set to "none", an {@code OffsetOutOfRangeException} will be thrown.
       * <p>
       * Note that, the seek offset won't change to the in-flight fetch request, it will take effect in next fetch request.
       * So, the consumer might wait for {@code fetch.max.wait.ms} before starting to fetch the records from desired offset.
       *
       * @param partition the TopicPartition on which the seek will be performed.
       * @param offset the next offset returned by poll().
       * @throws IllegalArgumentException if the provided offset is negative
       * @throws IllegalStateException if the provided TopicPartition is not assigned to this consumer
       */
      @Override
      public void seek(TopicPartition partition, long offset) {
          delegate.seek(partition, offset);
      }
  
      /**
       * Overrides the fetch offsets that the consumer will use on the next {@link #poll(Duration) poll(timeout)}. If this API
       * is invoked for the same partition more than once, the latest offset will be used on the next poll(). Note that
       * you may lose data if this API is arbitrarily used in the middle of consumption, to reset the fetch offsets. This
       * method allows for setting the leaderEpoch along with the desired offset.
       *
       * @throws IllegalArgumentException if the provided offset is negative
       * @throws IllegalStateException if the provided TopicPartition is not assigned to this consumer
       */
      @Override
      public void seek(TopicPartition partition, OffsetAndMetadata offsetAndMetadata) {
          delegate.seek(partition, offsetAndMetadata);
      }
  
      /**
       * Seek to the first offset for each of the given partitions. This function evaluates lazily, seeking to the
       * first offset in all partitions only when {@link #poll(Duration)} or {@link #position(TopicPartition)} are called.
       * If no partitions are provided, seek to the first offset for all of the currently assigned partitions.
       *
       * @throws IllegalArgumentException if {@code partitions} is {@code null}
       * @throws IllegalStateException if any of the provided partitions are not currently assigned to this consumer
       */
      @Override
      public void seekToBeginning(Collection<TopicPartition> partitions) {
          delegate.seekToBeginning(partitions);
      }
  
      /**
       * Seek to the last offset for each of the given partitions. This function evaluates lazily, seeking to the
       * final offset in all partitions only when {@link #poll(Duration)} or {@link #position(TopicPartition)} are called.
       * If no partitions are provided, seek to the final offset for all of the currently assigned partitions.
       * <p>
       * If {@code isolation.level=read_committed}, the end offset will be the Last Stable Offset, i.e., the offset
       * of the first message with an open transaction.
       *
       * @throws IllegalArgumentException if {@code partitions} is {@code null}
       * @throws IllegalStateException if any of the provided partitions are not currently assigned to this consumer
       */
      @Override
      public void seekToEnd(Collection<TopicPartition> partitions) {
          delegate.seekToEnd(partitions);
      }
  
      /**
       * Get the offset of the <i>next record</i> that will be fetched (if a record with that offset exists).
       * This method may issue a remote call to the server if there is no current position for the given partition.
       * <p>
       * This call will block until either the position could be determined or an unrecoverable error is
       * encountered (in which case it is thrown to the caller), or the timeout specified by {@code default.api.timeout.ms} expires
       * (in which case a {@link org.apache.kafka.common.errors.TimeoutException} is thrown to the caller).
       *
       * @param partition The partition to get the position for
       * @return The current position of the consumer (that is, the offset of the next record to be fetched)
       * @throws IllegalStateException if the provided TopicPartition is not assigned to this consumer
       * @throws org.apache.kafka.clients.consumer.InvalidOffsetException if no offset is currently defined for
       *             the partition
       * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
       *             function is called
       * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
       *             this function is called
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the
       *             configured groupId. See the exception for more details
       * @throws org.apache.kafka.common.errors.UnsupportedVersionException if the consumer attempts to fetch stable offsets
       *             when the broker doesn't support this feature
       * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors
       * @throws org.apache.kafka.common.errors.TimeoutException if the position cannot be determined before the
       *             timeout specified by {@code default.api.timeout.ms} expires
       */
      @Override
      public long position(TopicPartition partition) {
          return delegate.position(partition);
      }
  
      /**
       * Get the offset of the <i>next record</i> that will be fetched (if a record with that offset exists).
       * This method may issue a remote call to the server if there is no current position
       * for the given partition.
       * <p>
       * This call will block until the position can be determined, an unrecoverable error is
       * encountered (in which case it is thrown to the caller), or the timeout expires.
       *
       * @param partition The partition to get the position for
       * @param timeout The maximum amount of time to await determination of the current position
       * @return The current position of the consumer (that is, the offset of the next record to be fetched)
       * @throws IllegalStateException if the provided TopicPartition is not assigned to this consumer
       * @throws org.apache.kafka.clients.consumer.InvalidOffsetException if no offset is currently defined for
       *             the partition
       * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
       *             function is called
       * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
       *             this function is called
       * @throws org.apache.kafka.common.errors.TimeoutException if the position cannot be determined before the
       *             passed timeout expires
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the
       *             configured groupId. See the exception for more details
       * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors
       */
      @Override
      public long position(TopicPartition partition, final Duration timeout) {
          return delegate.position(partition, timeout);
      }
  
      /**
       * Get the last committed offsets for the given partitions (whether the commit happened by this process or
       * another). The returned offsets will be used as the position for the consumer in the event of a failure.
       * <p>
       * If any of the partitions requested do not exist, an exception would be thrown.
       * <p>
       * This call will do a remote call to get the latest committed offsets from the server, and will block until the
       * committed offsets are gotten successfully, an unrecoverable error is encountered (in which case it is thrown to
       * the caller), or the timeout specified by {@code default.api.timeout.ms} expires (in which case a
       * {@link org.apache.kafka.common.errors.TimeoutException} is thrown to the caller).
       *
       * @param partitions The partitions to check
       * @return The latest committed offsets for the given partitions; {@code null} will be returned for the
       *         partition if there is no such message.
       * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
       *             function is called
       * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
       *             this function is called
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the
       *             configured groupId. See the exception for more details
       * @throws org.apache.kafka.common.errors.UnsupportedVersionException if the consumer attempts to fetch stable offsets
       *             when the broker doesn't support this feature
       * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors
       * @throws org.apache.kafka.common.errors.TimeoutException if the committed offset cannot be found before
       *             the timeout specified by {@code default.api.timeout.ms} expires.
       */
      @Override
      public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition> partitions) {
          return delegate.committed(partitions);
      }
  
      /**
       * Get the last committed offsets for the given partitions (whether the commit happened by this process or
       * another). The returned offsets will be used as the position for the consumer in the event of a failure.
       * <p>
       * If any of the partitions requested do not exist, an exception would be thrown.
       * <p>
       * This call will block to do a remote call to get the latest committed offsets from the server.
       *
       * @param partitions The partitions to check
       * @param timeout  The maximum amount of time to await the latest committed offsets
       * @return The latest committed offsets for the given partitions; {@code null} will be returned for the
       *         partition if there is no such message.
       * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
       *             function is called
       * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
       *             this function is called
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the
       *             configured groupId. See the exception for more details
       * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors
       * @throws org.apache.kafka.common.errors.TimeoutException if the committed offset cannot be found before
       *             expiration of the timeout
       */
      @Override
      public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition> partitions, final Duration timeout) {
          return delegate.committed(partitions, timeout);
      }
  
      /**
       * Determines the client's unique client instance ID used for telemetry. This ID is unique to
       * this specific client instance and will not change after it is initially generated.
       * The ID is useful for correlating client operations with telemetry sent to the broker and
       * to its eventual monitoring destinations.
       * <p>
       * If telemetry is enabled, this will first require a connection to the cluster to generate
       * the unique client instance ID. This method waits up to {@code timeout} for the consumer
       * client to complete the request.
       * <p>
       * Client telemetry is controlled by the {@link ConsumerConfig#ENABLE_METRICS_PUSH_CONFIG}
       * configuration option.
       *
       * @param timeout The maximum time to wait for consumer client to determine its client instance ID.
       *                The value must be non-negative. Specifying a timeout of zero means do not
       *                wait for the initial request to complete if it hasn't already.
       * @throws InterruptException If the thread is interrupted while blocked.
       * @throws KafkaException If an unexpected error occurs while trying to determine the client
       *                        instance ID, though this error does not necessarily imply the
       *                        consumer client is otherwise unusable.
       * @throws IllegalArgumentException If the {@code timeout} is negative.
       * @throws IllegalStateException If telemetry is not enabled ie, config `{@code enable.metrics.push}`
       *                               is set to `{@code false}`.
       * @return The client's assigned instance id used for metrics collection.
       */
      @Override
      public Uuid clientInstanceId(Duration timeout) {
          return delegate.clientInstanceId(timeout);
      }
  
    /**
       * Get the metrics kept by the consumer
       */
      @Override
      public Map<MetricName, ? extends Metric> metrics() {
          return delegate.metrics();
      }
  
      /**
       * Get metadata about the partitions for a given topic. This method will issue a remote call to the server if it
       * does not already have any metadata about the given topic.
       *
       * @param topic The topic to get partition metadata for
       *
       * @return The list of partitions, which will be empty when the given topic is not found
       * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
       *             function is called
       * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
       *             this function is called
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the specified topic. See the exception for more details
       * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors
       * @throws org.apache.kafka.common.errors.TimeoutException if the offset metadata could not be fetched before
       *         the amount of time allocated by {@code default.api.timeout.ms} expires.
       */
      @Override
      public List<PartitionInfo> partitionsFor(String topic) {
          return delegate.partitionsFor(topic);
      }
  
      /**
       * Get metadata about the partitions for a given topic. This method will issue a remote call to the server if it
       * does not already have any metadata about the given topic.
       *
       * @param topic The topic to get partition metadata for
       * @param timeout The maximum of time to await topic metadata
       *
       * @return The list of partitions, which will be empty when the given topic is not found
       * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
       *             function is called
       * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
       *             this function is called
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the specified topic. See
       *             the exception for more details
       * @throws org.apache.kafka.common.errors.TimeoutException if topic metadata cannot be fetched before expiration
       *             of the passed timeout
       * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors
       */
      @Override
      public List<PartitionInfo> partitionsFor(String topic, Duration timeout) {
          return delegate.partitionsFor(topic, timeout);
      }
  
      /**
       * Get metadata about partitions for all topics that the user is authorized to view. This method will issue a
       * remote call to the server.
  
       * @return The map of topics and its partitions
       *
       * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
       *             function is called
       * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
       *             this function is called
       * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors
       * @throws org.apache.kafka.common.errors.TimeoutException if the offset metadata could not be fetched before
       *         the amount of time allocated by {@code default.api.timeout.ms} expires.
       */
      @Override
      public Map<String, List<PartitionInfo>> listTopics() {
          return delegate.listTopics();
      }
  
      /**
       * Get metadata about partitions for all topics that the user is authorized to view. This method will issue a
       * remote call to the server.
       *
       * @param timeout The maximum time this operation will block to fetch topic metadata
       *
       * @return The map of topics and its partitions
       * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this
       *             function is called
       * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while
       *             this function is called
       * @throws org.apache.kafka.common.errors.TimeoutException if the topic metadata could not be fetched before
       *             expiration of the passed timeout
       * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors
       */
      @Override
      public Map<String, List<PartitionInfo>> listTopics(Duration timeout) {
          return delegate.listTopics(timeout);
      }
  
      /**
       * Suspend fetching from the requested partitions. Future calls to {@link #poll(Duration)} will not return
       * any records from these partitions until they have been resumed using {@link #resume(Collection)}.
       * Note that this method does not affect partition subscription. In particular, it does not cause a group
       * rebalance when automatic assignment is used.
       *
       * Note: Rebalance will not preserve the pause/resume state.
       * @param partitions The partitions which should be paused
       * @throws IllegalStateException if any of the provided partitions are not currently assigned to this consumer
       */
      @Override
      public void pause(Collection<TopicPartition> partitions) {
          delegate.pause(partitions);
      }
  
      /**
       * Resume specified partitions which have been paused with {@link #pause(Collection)}. New calls to
       * {@link #poll(Duration)} will return records from these partitions if there are any to be fetched.
       * If the partitions were not previously paused, this method is a no-op.
       * @param partitions The partitions which should be resumed
       * @throws IllegalStateException if any of the provided partitions are not currently assigned to this consumer
       */
      @Override
      public void resume(Collection<TopicPartition> partitions) {
          delegate.resume(partitions);
      }
  
      /**
       * Add the provided application metric for subscription.
       * This metric will be added to this client's metrics
       * that are available for subscription and sent as
       * telemetry data to the broker.
       * The provided metric must map to an OTLP metric data point
       * type in the OpenTelemetry v1 metrics protobuf message types.
       * Specifically, the metric should be one of the following:
       * <ul>
       *  <li>
       *     `Sum`: Monotonic total count meter (Counter). Suitable for metrics like total number of X, e.g., total bytes sent.
       *  </li>
       *  <li>
       *     `Gauge`: Non-monotonic current value meter (UpDownCounter). Suitable for metrics like current value of Y, e.g., current queue count.
       *  </li>
       * </ul>
       * Metrics not matching these types are silently ignored.
       * Executing this method for a previously registered metric is a benign operation and results in updating that metrics entry.
       *
       * @param metric The application metric to register
       */
      @Override
      public void registerMetricForSubscription(KafkaMetric metric) {
          delegate.registerMetricForSubscription(metric);
      }
  
      /**
       * Remove the provided application metric for subscription.
       * This metric is removed from this client's metrics
       * and will not be available for subscription any longer.
       * Executing this method with a metric that has not been registered is a
       * benign operation and does not result in any action taken (no-op).
       *
       * @param metric The application metric to remove
       */
      @Override
      public void unregisterMetricFromSubscription(KafkaMetric metric) {
          delegate.unregisterMetricFromSubscription(metric);
      }
  
      /**
       * Get the set of partitions that were previously paused by a call to {@link #pause(Collection)}.
       *
       * @return The set of paused partitions
       */
      @Override
      public Set<TopicPartition> paused() {
          return delegate.paused();
      }
  
      /**
       * Look up the offsets for the given partitions by timestamp. The returned offset for each partition is the
       * earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition.
       *
       * This is a blocking call. The consumer does not have to be assigned the partitions.
       * If the message format version in a partition is before 0.10.0, i.e. the messages do not have timestamps, null
       * will be returned for that partition.
       *
       * @param timestampsToSearch the mapping from partition to the timestamp to look up.
       *
       * @return a mapping from partition to the timestamp and offset of the first message with timestamp greater
       *         than or equal to the target timestamp. If the timestamp and offset for a specific partition cannot be found within
       *         the default timeout, and no corresponding message exists, the entry in the returned map will be {@code null}
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic(s). See the exception for more details
       * @throws IllegalArgumentException if the target timestamp is negative
       * @throws org.apache.kafka.common.errors.TimeoutException if the offset metadata could not be fetched before
       *         the amount of time allocated by {@code default.api.timeout.ms} expires.
       * @throws org.apache.kafka.common.errors.UnsupportedVersionException if the broker does not support looking up
       *         the offsets by timestamp
       */
      @Override
      public Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch) {
          return delegate.offsetsForTimes(timestampsToSearch);
      }
  
      /**
       * Look up the offsets for the given partitions by timestamp. The returned offset for each partition is the
       * earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition.
       *
       * This is a blocking call. The consumer does not have to be assigned the partitions.
       * If the message format version in a partition is before 0.10.0, i.e. the messages do not have timestamps, null
       * will be returned for that partition.
       *
       * @param timestampsToSearch the mapping from partition to the timestamp to look up.
       * @param timeout The maximum amount of time to await retrieval of the offsets
       *
       * @return a mapping from partition to the timestamp and offset of the first message with timestamp greater
       *         than or equal to the target timestamp. If the timestamp and offset for a specific partition cannot be found within
       *         timeout, and no corresponding message exists, the entry in the returned map will be {@code null}
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic(s). See the exception for more details
       * @throws IllegalArgumentException if the target timestamp is negative
       * @throws org.apache.kafka.common.errors.TimeoutException if the offset metadata could not be fetched before
       *         expiration of the passed timeout
       * @throws org.apache.kafka.common.errors.UnsupportedVersionException if the broker does not support looking up
       *         the offsets by timestamp
       */
      @Override
      public Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch, Duration timeout) {
          return delegate.offsetsForTimes(timestampsToSearch, timeout);
      }
  
      /**
       * Get the first offset for the given partitions.
       * <p>
       * This method does not change the current consumer position of the partitions.
       *
       * @see #seekToBeginning(Collection)
       *
       * @param partitions the partitions to get the earliest offsets.
       * @return The earliest available offsets for the given partitions
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic(s). See the exception for more details
       * @throws org.apache.kafka.common.errors.TimeoutException if the offset metadata could not be fetched before
       *         expiration of the configured {@code default.api.timeout.ms}
       */
      @Override
      public Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions) {
          return delegate.beginningOffsets(partitions);
      }
  
      /**
       * Get the first offset for the given partitions.
       * <p>
       * This method does not change the current consumer position of the partitions.
       *
       * @see #seekToBeginning(Collection)
       *
       * @param partitions the partitions to get the earliest offsets
       * @param timeout The maximum amount of time to await retrieval of the beginning offsets
       *
       * @return The earliest available offsets for the given partitions, and it will return empty map if zero timeout is provided
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic(s). See the exception for more details
       * @throws org.apache.kafka.common.errors.TimeoutException if the offset metadata could not be fetched before
       *         expiration of the passed timeout
       */
      @Override
      public Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions, Duration timeout) {
          return delegate.beginningOffsets(partitions, timeout);
      }
  
      /**
       * Get the end offsets for the given partitions. In the default {@code read_uncommitted} isolation level, the end
       * offset is the high watermark (that is, the offset of the last successfully replicated message plus one). For
       * {@code read_committed} consumers, the end offset is the last stable offset (LSO), which is the minimum of
       * the high watermark and the smallest offset of any open transaction. Finally, if the partition has never been
       * written to, the end offset is 0.
       *
       * <p>
       * This method does not change the current consumer position of the partitions.
       *
       * @see #seekToEnd(Collection)
       *
       * @param partitions the partitions to get the end offsets.
       * @return The end offsets for the given partitions.
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic(s). See the exception for more details
       * @throws org.apache.kafka.common.errors.TimeoutException if the offset metadata could not be fetched before
       *         the amount of time allocated by {@code default.api.timeout.ms} expires
       */
      @Override
      public Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions) {
          return delegate.endOffsets(partitions);
      }
  
      /**
       * Get the end offsets for the given partitions. In the default {@code read_uncommitted} isolation level, the end
       * offset is the high watermark (that is, the offset of the last successfully replicated message plus one). For
       * {@code read_committed} consumers, the end offset is the last stable offset (LSO), which is the minimum of
       * the high watermark and the smallest offset of any open transaction. Finally, if the partition has never been
       * written to, the end offset is 0.
       *
       * <p>
       * This method does not change the current consumer position of the partitions.
       *
       * @see #seekToEnd(Collection)
       *
       * @param partitions the partitions to get the end offsets.
       * @param timeout The maximum amount of time to await retrieval of the end offsets
       *
       * @return The end offsets for the given partitions, and it will return empty map if zero timeout is provided
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic(s). See the exception for more details
       * @throws org.apache.kafka.common.errors.TimeoutException if the offsets could not be fetched before
       *         expiration of the passed timeout
       */
      @Override
      public Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> partitions, Duration timeout) {
          return delegate.endOffsets(partitions, timeout);
      }
  
      /**
       * Get the consumer's current lag on the partition. Returns an "empty" {@link OptionalLong} if the lag is not known,
       * for example if there is no position yet, or if the end offset is not known yet.
       *
       * <p>
       * This method uses locally cached metadata. If the log end offset is not known yet, it triggers a request to fetch
       * the log end offset, but returns immediately.
       *
       * @param topicPartition The partition to get the lag for.
       *
       * @return This {@code Consumer} instance's current lag for the given partition.
       *
       * @throws IllegalStateException if the {@code topicPartition} is not assigned
       */
      @Override
      public OptionalLong currentLag(TopicPartition topicPartition) {
          return delegate.currentLag(topicPartition);
      }
  
      /**
       * Return the current group metadata associated with this consumer.
       *
       * @return consumer group metadata
       * @throws org.apache.kafka.common.errors.InvalidGroupIdException if consumer does not have a group
       */
      @Override
      public ConsumerGroupMetadata groupMetadata() {
          return delegate.groupMetadata();
      }
  
      /**
       * Alert the consumer to trigger a new rebalance by rejoining the group. This is a nonblocking call that forces
       * the consumer to trigger a new rebalance on the next {@link #poll(Duration)} call. Note that this API does not
       * itself initiate the rebalance, so you must still call {@link #poll(Duration)}. If a rebalance is already in
       * progress this call will be a no-op. If you wish to force an additional rebalance you must complete the current
       * one by calling poll before retrying this API.
       * <p>
       * You do not need to call this during normal processing, as the consumer group will manage itself
       * automatically and rebalance when necessary. However there may be situations where the application wishes to
       * trigger a rebalance that would otherwise not occur. For example, if some condition external and invisible to
       * the Consumer and its group changes in a way that would affect the userdata encoded in the
       * {@link org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.Subscription Subscription}, the Consumer
       * will not be notified and no rebalance will occur. This API can be used to force the group to rebalance so that
       * the assignor can perform a partition reassignment based on the latest userdata. If your assignor does not use
       * this userdata, or you do not use a custom
       * {@link org.apache.kafka.clients.consumer.ConsumerPartitionAssignor ConsumerPartitionAssignor}, you should not
       * use this API.
       *
       * @param reason The reason why the new rebalance is needed.
       *
       * @throws java.lang.IllegalStateException if the consumer does not use group subscription
       */
      @Override
      public void enforceRebalance(final String reason) {
          delegate.enforceRebalance(reason);
      }
  
      /**
       * @see #enforceRebalance(String)
       */
      @Override
      public void enforceRebalance() {
          delegate.enforceRebalance();
      }
  
      /**
       * Close the consumer with {@link CloseOptions.GroupMembershipOperation#DEFAULT default leave group behavior},
       * waiting for up to the default timeout of 30 seconds for any needed cleanup.
       * If auto-commit is enabled, this will commit the current offsets if possible within the default
       * timeout. See {@link #close(CloseOptions)} for details. Note that {@link #wakeup()}
       * cannot be used to interrupt close.
       *
       * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted
       *             before or while this function is called
       * @throws org.apache.kafka.common.KafkaException for any other error during close
       */
      @Override
      public void close() {
          delegate.close();
      }
  
      /**
       * This method has been deprecated since Kafka 4.1 and should use {@link KafkaConsumer#close(CloseOptions)} instead.
       * <p>
       * Close the consumer with {@link CloseOptions.GroupMembershipOperation#DEFAULT default leave group behavior}
       * cleanly within the specified timeout. This method waits up to
       * {@code timeout} for the consumer to complete pending commits and maybe leave the group (if the member is dynamic).
       * If auto-commit is enabled, this will commit the current offsets if possible within the
       * timeout. If the consumer is unable to complete offset commits and to gracefully leave the group (if applicable)
       * before the timeout expires, the consumer is force closed. Note that {@link #wakeup()} cannot be
       * used to interrupt close.
       * <p>
       * The actual maximum wait time is bounded by the {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} setting, which
       * only applies to operations performed with the broker (coordinator-related requests and
       * fetch sessions). Even if a larger timeout is specified, the consumer will not wait longer than
       * {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} for these requests to complete during the close operation.
       * Note that the execution time of callbacks (such as {@link OffsetCommitCallback} and
       * {@link ConsumerRebalanceListener}) does not consume time from the close timeout.
       *
       * @param timeout The maximum time to wait for consumer to close gracefully. The value must be
       *                non-negative. Specifying a timeout of zero means do not wait for pending requests to complete.
       *
       * @throws IllegalArgumentException If the {@code timeout} is negative.
       * @throws InterruptException If the thread is interrupted before or while this function is called
       * @throws org.apache.kafka.common.KafkaException for any other error during close
       */
      @Deprecated(since = "4.1")
      @Override
      public void close(Duration timeout) {
          delegate.close(timeout);
      }
  
      /**
       * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
       * {@link CloseOptions.GroupMembershipOperation leave group behavior}.
       * If no timeout is specified, the default timeout of 30 seconds is used.
       * If no leave group behavior is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
       * leave group behavior} is used.
       * <p>
       * This method waits up to the timeout for the consumer to complete pending commits and maybe leave the group,
       * depending on the specified leave group behavior.
       * If auto-commit is enabled, this will commit the current offsets if possible within the
       * timeout. If the consumer is unable to complete offset commits and to gracefully leave the group (if applicable)
       * before the timeout expires, the consumer is force closed. Note that {@link #wakeup()} cannot be
       * used to interrupt close.
       * <p>
       * The actual maximum wait time is bounded by the {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} setting, which
       * only applies to operations performed with the broker (coordinator-related requests and
       * fetch sessions). Even if a larger timeout is specified, the consumer will not wait longer than
       * {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} for these requests to complete during the close operation.
       * Note that the execution time of callbacks (such as {@link OffsetCommitCallback} and
       * {@link ConsumerRebalanceListener}) does not consume time from the close timeout.
       *
       * @param option see {@link CloseOptions}; cannot be {@code null}
       */
      @Override
      public void close(CloseOptions option) {
          delegate.close(option);
      }
  
      /**
       * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
       * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
       * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
       */
      @Override
      public void wakeup() {
          delegate.wakeup();
      }
  
      // Functions below are for testing only
      String clientId() {
          return delegate.clientId();
      }
  
      Metrics metricsRegistry() {
          return delegate.metricsRegistry();
      }
  
      KafkaConsumerMetrics kafkaConsumerMetrics() {
          return delegate.kafkaConsumerMetrics();
      }
  
      boolean updateAssignmentMetadataIfNeeded(final Timer timer) {
          return delegate.updateAssignmentMetadataIfNeeded(timer);
      }
  }
  