class Fetcher {

    public void updateFetchPositions(Set<TopicPartition> partitions) {
        // reset the fetch position to the committed position
        for (TopicPartition tp : partitions) {
            if (!subscriptions.isAssigned(tp) || subscriptions.isFetchable(tp)) {// 没分配或者没暂停之类的才能update
                continue;
            }

            // TODO: If there are several offsets to reset, we could submit offset requests in parallel
            if (subscriptions.isOffsetResetNeeded(tp)) {// resetStrategy != null; 重设策略不为空
                resetOffset(tp);
            } else if (subscriptions.committed(tp) == null) {
                // there's no committed position, so we need to reset with the default strategy
                subscriptions.needOffsetReset(tp);
                resetOffset(tp);
            } else {
                // 如果没有重置策略则简单将commit设置为offset
                long committed = subscriptions.committed(tp)
                                              .offset();
                log.debug("Resetting offset for partition {} to the committed offset {}", tp, committed);
                subscriptions.seek(tp, committed);
            }
        }
    }

    public  Fetcher(ConsumerNetworkClient client,
        int minBytes,
        int maxWaitMs,
        int fetchSize,
        int maxPollRecords,
        boolean checkCrcs,
        Deserializer<K> keyDeserializer,
        Deserializer<V> valueDeserializer,
        Metadata metadata,
        SubscriptionState subscriptions,
        Metrics metrics,
        String metricGrpPrefix,
        Time time,
        long retryBackoffMs);

}