kafka消费者--offset提交

public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets, final Duration timeout) {
//加锁
acquireAndEnsureOpen();
try {
maybeThrowInvalidGroupIdException();
offsets.forEach(this::updateLastSeenEpochIfNewer);
//提交offset值
if (!coordinator.commitOffsetsSync(new HashMap<>(offsets), time.timer(timeout))) {
throw new TimeoutException("Timeout of " + timeout.toMillis() + "ms expired before successfully " +
"committing offsets " + offsets);
}
} finally {
//释放锁
release();
}
}
 

public boolean commitOffsetsSync(Map<TopicPartition, OffsetAndMetadata> offsets, Timer timer) {
//异步提交完成offset回调callback
invokeCompletedOffsetCommitCallbacks();
//如果没有offset要提交直接返回
if (offsets.isEmpty())
return true;
//发送主循环体
do {
//协调者未知或者连接不上直接返回失败
if (coordinatorUnknown() && !ensureCoordinatorReady(timer)) {
return false;
}
//组装发送请求并唤醒客服端
RequestFuture<Void> future = sendOffsetCommitRequest(offsets);
//发送并获取结果
client.poll(future, timer);
//如果poll后获取到之前异步提交的offset，可以直接回调callback
invokeCompletedOffsetCommitCallbacks();
//如果发送成功，返回成功结果
if (future.succeeded()) {
if (interceptors != null)
interceptors.onCommit(offsets);
return true;
}
//如果失败且不能重试的异常，直接抛出异常
if (future.failed() && !future.isRetriable())
throw future.exception();
//如果可以重试的异常，就休眠一段时间进行重试
timer.sleep(rebalanceConfig.retryBackoffMs);
} while (timer.notExpired());
//如果超时还没提交上去就返回失败
return false;
}

void invokeCompletedOffsetCommitCallbacks() {
if (asyncCommitFenced.get()) {
throw new FencedInstanceIdException("Get fenced exception for group.instance.id "
+ rebalanceConfig.groupInstanceId.orElse("unset_instance_id")
+ ", current member.id is " + memberId());
}
while (true) {
//从队列里面处理已经完成提交的offset进行回调
OffsetCommitCompletion completion = completedOffsetCommits.poll();
//当队列为空时退出
if (completion == null) {
break;
}
completion.invoke();
}
}

RequestFuture<Void> sendOffsetCommitRequest(final Map<TopicPartition, OffsetAndMetadata> offsets) {
//offset 为空，直接返回
if (offsets.isEmpty())
return RequestFuture.voidSuccess();
//获取协调者
Node coordinator = checkAndGetCoordinator();
if (coordinator == null)
return RequestFuture.coordinatorNotAvailable();
 
// 组装offset的数据格式
Map<String, OffsetCommitRequestData.OffsetCommitRequestTopic> requestTopicDataMap = new HashMap<>();
for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) {
TopicPartition topicPartition = entry.getKey();
OffsetAndMetadata offsetAndMetadata = entry.getValue();
if (offsetAndMetadata.offset() < 0) {
return RequestFuture.failure(new IllegalArgumentException("Invalid offset: " + offsetAndMetadata.offset()));
}
 
OffsetCommitRequestData.OffsetCommitRequestTopic topic = requestTopicDataMap
.getOrDefault(topicPartition.topic(),
new OffsetCommitRequestData.OffsetCommitRequestTopic()
.setName(topicPartition.topic())
);
 
topic.partitions().add(new OffsetCommitRequestData.OffsetCommitRequestPartition()
.setPartitionIndex(topicPartition.partition())
.setCommittedOffset(offsetAndMetadata.offset())
.setCommittedLeaderEpoch(offsetAndMetadata.leaderEpoch().orElse(RecordBatch.NO_PARTITION_LEADER_EPOCH))
.setCommittedMetadata(offsetAndMetadata.metadata())
);
requestTopicDataMap.put(topicPartition.topic(), topic);
}
//获取版本号
final Generation generation;
if (subscriptions.hasAutoAssignedPartitions()) {
generation = generationIfStable();
//...异常处理并返回
} else {
generation = Generation.NO_GENERATION;
}
//组装发送参数
OffsetCommitRequest.Builder builder = new OffsetCommitRequest.Builder(
new OffsetCommitRequestData()
.setGroupId(this.rebalanceConfig.groupId)
.setGenerationId(generation.generationId)
.setMemberId(generation.memberId)
.setGroupInstanceId(rebalanceConfig.groupInstanceId.orElse(null))
.setTopics(new ArrayList<>(requestTopicDataMap.values()))
);
//发送到unset队列等待客服端发送
return client.send(coordinator, builder)
.compose(new OffsetCommitResponseHandler(offsets, generation));
}

public void handle(OffsetCommitResponse commitResponse, RequestFuture<Void> future) {
//未认证的topic列表
Set<String> unauthorizedTopics = new HashSet<>();
//对每一个结果进行处理，主要根据不同的异常进行不同的处理
for (OffsetCommitResponseData.OffsetCommitResponseTopic topic : commitResponse.data().topics()) {
for (OffsetCommitResponseData.OffsetCommitResponsePartition partition : topic.partitions()) {
TopicPartition tp = new TopicPartition(topic.name(), partition.partitionIndex());
OffsetAndMetadata offsetAndMetadata = this.offsets.get(tp);
 
long offset = offsetAndMetadata.offset();
 
Errors error = Errors.forCode(partition.errorCode());
if (error == Errors.NONE) {
log.debug("Committed offset {} for partition {}", offset, tp);
} else {
//异常处理
//...
}
}
}
//抛出未认证异常
if (!unauthorizedTopics.isEmpty()) {
future.raise(new TopicAuthorizationException(unauthorizedTopics));
} else {
//返回成功
future.complete(null);
}
}

private void doCommitOffsetsAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, final OffsetCommitCallback callback) {
//发送
RequestFuture<Void> future = sendOffsetCommitRequest(offsets);
//创建callback
final OffsetCommitCallback cb = callback == null ? defaultOffsetCommitCallback : callback;
//添加callback的监听器
future.addListener(new RequestFutureListener<Void>() {
@Override
public void onSuccess(Void value) {
if (interceptors != null)
interceptors.onCommit(offsets);
//发送成功，添加到callback中
completedOffsetCommits.add(new OffsetCommitCompletion(cb, offsets, null));
}
 
@Override
public void onFailure(RuntimeException e) {
Exception commitException = e;
 
if (e instanceof RetriableException) {
commitException = new RetriableCommitFailedException(e);
}
//发送失败，也添加到callback中
completedOffsetCommits.add(new OffsetCommitCompletion(cb, offsets, commitException));
if (commitException instanceof FencedInstanceIdException) {
asyncCommitFenced.set(true);
}
}
});
}

public void maybeAutoCommitOffsetsAsync(long now) {
//是否开启自动提交
if (autoCommitEnabled) {
nextAutoCommitTimer.update(now);
if (nextAutoCommitTimer.isExpired()) {
nextAutoCommitTimer.reset(autoCommitIntervalMs);
//执行自动提交
doAutoCommitOffsetsAsync();
}
}
}
private void doAutoCommitOffsetsAsync() {
Map<TopicPartition, OffsetAndMetadata> allConsumedOffsets = subscriptions.allConsumed();
log.debug("Sending asynchronous auto-commit of offsets {}", allConsumedOffsets);
//异步提交offset
commitOffsetsAsync(allConsumedOffsets, (offsets, exception) -> {
if (exception != null) {
if (exception instanceof RetriableCommitFailedException) {
log.debug("Asynchronous auto-commit of offsets {} failed due to retriable error: {}", offsets,
exception);
nextAutoCommitTimer.updateAndReset(rebalanceConfig.retryBackoffMs);
} else {
log.warn("Asynchronous auto-commit of offsets {} failed: {}", offsets, exception.getMessage());
}
} else {
log.debug("Completed asynchronous auto-commit of offsets {}", offsets);
}
});
}