public ConsumerRecords<K, V> poll(long timeout) {

    acquire();

    try {

        if (timeout < 0)

            throw new IllegalArgumentException("Timeout must not be negative");

        // 如果没有任何订阅，抛出异常

        if (this.subscriptions.hasNoSubscriptionOrUserAssignment())

            throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");

        // 一直poll新数据直到超时

        long start = time.milliseconds();

        // 距离超时还剩余多少时间

        long remaining = timeout;

        do {

            // 获取数据，如果自动提交，则进行偏移量自动提交，如果设置offset重置，则进行offset重置

            Map<TopicPartition, List<ConsumerRecord<K, V>>> records = pollOnce(remaining);

            if (!records.isEmpty()) {

                // 再返回结果之前，我们可以进行下一轮的fetch请求，避免阻塞等待

                fetcher.sendFetches();

                client.pollNoWakeup();

                // 如果有拦截器进行拦截，没有直接返回

                if (this.interceptors == null)

                    return new ConsumerRecords<>(records);

                else

                    return this.interceptors.onConsume(new ConsumerRecords<>(records));

            }

            long elapsed = time.milliseconds() - start;

            remaining = timeout - elapsed;

        } while (remaining > 0);

        return ConsumerRecords.empty();

    } finally {

        release();

    }

}