1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
private void doCommitOffsetsAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, final OffsetCommitCallback callback) {
    // 将needsFetchCommittedOffsets置为true
    this.subscriptions.needRefreshCommits();
    // 发送OffsetCommitRequest请求
    RequestFuture<Void> future = sendOffsetCommitRequest(offsets);
    final OffsetCommitCallback cb = callback == null ? defaultOffsetCommitCallback : callback;
    // 对RequestFuture添加监听器，如果成功则completedOffsetCommits添加OffsetCommitCompletion
    // 如果失败也往completedOffsetCommits添加OffsetCommitCompletion，但是会带有异常
    future.addListener(new RequestFutureListener<Void>() {
        @Override
        public void onSuccess(Void value) {
            if (interceptors != null)
                interceptors.onCommit(offsets);

            completedOffsetCommits.add(new OffsetCommitCompletion(cb, offsets, null));
        }

        @Override
        public void onFailure(RuntimeException e) {
            Exception commitException = e;

            if (e instanceof RetriableException)
                commitException = new RetriableCommitFailedException(e);

            completedOffsetCommits.add(new OffsetCommitCompletion(cb, offsets, commitException));
        }
    });
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
private RequestFuture<Void> sendOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
    if (offsets.isEmpty())
        return RequestFuture.voidSuccess();
    // 获取GroupCoordinator
    Node coordinator = coordinator();
    if (coordinator == null)
        return RequestFuture.coordinatorNotAvailable();

    // 遍历offsets,并且往offsetData里put数据
    Map<TopicPartition, OffsetCommitRequest.PartitionData> offsetData = new HashMap<>(offsets.size());
    for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {
        OffsetAndMetadata offsetAndMetadata = entry.getValue();
        offsetData.put(entry.getKey(), new OffsetCommitRequest.PartitionData(
                offsetAndMetadata.offset(), offsetAndMetadata.metadata()));
    }
    // 获取generation信息
    final Generation generation;
    if (subscriptions.partitionsAutoAssigned())
        generation = generation();
    else
        generation = Generation.NO_GENERATION;

    // if the generation is null, we are not part of an active group (and we expect to be).
    // the only thing we can do is fail the commit and let the user rejoin the group in poll()
    if (generation == null)
        return RequestFuture.failure(new CommitFailedException());
    // 构建OffsetCommitRequest
    OffsetCommitRequest req = new OffsetCommitRequest(
            this.groupId,
            generation.generationId,
            generation.memberId,
            OffsetCommitRequest.DEFAULT_RETENTION_TIME,
            offsetData);

    log.trace("Sending offset-commit request with {} to coordinator {} for group {}", offsets, coordinator, groupId);
    // 将请求放到unsent集合等待被发送
    return client.send(coordinator, ApiKeys.OFFSET_COMMIT, req)
            .compose(new OffsetCommitResponseHandler(offsets));
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public Map<TopicPartition, OffsetAndMetadata> fetchCommittedOffsets(Set<TopicPartition> partitions) {
    while (true) {
        // 确认已经准备好接受请求，否则一直阻塞
        ensureCoordinatorReady();
        // 发送OffsetFetchRequest请求
        RequestFuture<Map<TopicPartition, OffsetAndMetadata>> future = sendOffsetFetchRequest(partitions);
        // 阻塞发送OffsetFetchRequest请求
        client.poll(future);
        if (future.succeeded())
            return future.value(); // 返回从服务器端得到的offset的值
        if (!future.isRetriable())
            throw future.exception();
        time.sleep(retryBackoffMs);// 否则等待重试
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
private RequestFuture<Map<TopicPartition, OffsetAndMetadata>> sendOffsetFetchRequest(Set<TopicPartition> partitions) {
    // 获取GroupCoordinator
    Node coordinator = coordinator();
    if (coordinator == null)
        return RequestFuture.coordinatorNotAvailable();

    log.debug("Group {} fetching committed offsets for partitions: {}", groupId, partitions);
    // 构建OffsetFetchRequest请求
    OffsetFetchRequest request = new OffsetFetchRequest(this.groupId, new ArrayList<>(partitions));

    // 发送请求到unsent请求队列，等待被发送
    return client.send(coordinator, ApiKeys.OFFSET_FETCH, request)
            .compose(new OffsetFetchResponseHandler());
}

1
处理OffsetFetchResponse：在返回RequestFuture的时候，会通过OffsetFetchResponseHandler对OffsetFetchResponse做一些处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
    public void handle(OffsetFetchResponse response, RequestFuture<Map<TopicPartition, OffsetAndMetadata>> future) {
        // 构造一个<TopicPartition, OffsetAndMetadata> 集合offsets
        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(response.responseData().size());
        // 遍历返回的相应的数据
        for (Map.Entry<TopicPartition, OffsetFetchResponse.PartitionData> entry : response.responseData().entrySet()) {
            TopicPartition tp = entry.getKey();
            OffsetFetchResponse.PartitionData data = entry.getValue();
            if (data.hasError()) {
                Errors error = Errors.forCode(data.errorCode);
                log.debug("Group {} failed to fetch offset for partition {}: {}", groupId, tp, error.message());

                if (error == Errors.GROUP_LOAD_IN_PROGRESS) {
                    // just retry
                    future.raise(error);
                } else if (error == Errors.NOT_COORDINATOR_FOR_GROUP) {
                    // re-discover the coordinator and retry
                    coordinatorDead();
                    future.raise(error);
                } else if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
                    future.raise(new KafkaException("Partition " + tp + " may not exist or user may not have Describe access to topic"));
                } else {
                    future.raise(new KafkaException("Unexpected error in fetch offset response: " + error.message()));
                }
                return;
            } else if (data.offset >= 0) {// 如果没有错误则把数据放到offsets集合里
                // record the position with the offset (-1 indicates no committed offset to fetch)
                offsets.put(tp, new OffsetAndMetadata(data.offset, data.metadata));
            } else {
                log.debug("Group {} has no committed offset for partition {}", groupId, tp);
            }
        }
        // 传播offset集合最终会通过fetchCommitedOffsets取到offset的值
        future.complete(offsets);
    }
}