class ConsumerCoordinator extends AbstractCoordinator {

    public void refreshCommittedOffsetsIfNeeded() {
        if (subscriptions.refreshCommitsNeeded()) {
            // 发送offsetFetch】Request，并处理OffsetFetchResponse响应，返回最近提交的offset集合
            Map<TopicPartition, OffsetAndMetadata> offsets = fetchCommittedOffsets(subscriptions.assignedPartitions());

            for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {
                TopicPartition tp = entry.getKey();
                // verify assignment is still active
                if (subscriptions.isAssigned(tp)) {
                    this.subscriptions.committed(tp, entry.getValue());
                }
            }
            this.subscriptions.commitsRefreshed();
        }
    }

    public  ConsumerCoordinator(ConsumerNetworkClient client,
        String groupId,
        int sessionTimeoutMs,
        int heartbeatIntervalMs,
        List<PartitionAssignor> assignors,
        Metadata metadata,
        SubscriptionState subscriptions,
        Metrics metrics,
        String metricGrpPrefix,
        Time time,
        long retryBackoffMs,
        OffsetCommitCallback defaultOffsetCommitCallback,
        boolean autoCommitEnabled,
        long autoCommitIntervalMs,
        ConsumerInterceptors<?, ?> interceptors,
        boolean excludeInternalTopics);

}