《Java 消息中间件·可靠消息与消费治理》12.2 异步提交的陷阱

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
    for (ConsumerRecord<String, String> r : records) {
        handle(r);                       // 业务处理
    }
    consumer.commitAsync();              // 处理完就提交，不阻塞
}

This is an asynchronous call and will not block. Any errors encountered are either passed to the callback
(if provided) or discarded.

Offsets committed through multiple calls to this API are guaranteed to be sent in the same order as
the invocations. Corresponding commit callbacks are also invoked in the same order. Additionally note that
offsets committed through this API are guaranteed to complete before a subsequent call to commitSync() returns.

private class DefaultOffsetCommitCallback implements OffsetCommitCallback {
    @Override
    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
        if (exception != null)
            log.error("Offset commit with offsets {} failed", offsets, exception);
    }
}

public boolean poll(Timer timer, boolean waitForJoinGroup) {
    invokeCompletedOffsetCommitCallbacks();
    // ... 之后才是心跳、协调器、拉取
}

[kafka-coordinator-heartbeat-thread | libraryd-async-kick-90718-A] WARN
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator -
consumer poll timeout has expired. This means the time between subsequent calls to poll() was longer than
the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time
processing messages.

[kick] A polled=10 firstOffset=0 nextOffset=10 at t=0
[kick] commitAsync(no callback) returned, threwSynchronously=false

[main] ERROR org.apache.kafka.clients.consumer.internals.ConsumerCoordinator -
Offset commit with offsets {library.loan.created.async.90718-0=OffsetAndMetadata{offset=10, ...}} failed
org.apache.kafka.clients.consumer.CommitFailedException: Offset commit cannot be completed since the
consumer is not part of an active group for auto partition assignment; it is likely that the consumer
was kicked out of the group.

[kick] group committed after failure={} (none)
[kick] A processed=10, committedOffset=none, exceptionReachedCaller=false
[kick] drive poll returned 10 records (callback queue drained)

[kick-sync] commitSync threw org.apache.kafka.clients.consumer.CommitFailedException:
Offset commit cannot be completed since the consumer is not part of an active group for auto partition
assignment; it is likely that the consumer was kicked out of the group.
[kick-sync] committed after failure={} (none)

ConsumerRecords<String, String> batch = consumer.poll(Duration.ofMillis(200));
for (ConsumerRecord<String, String> r : batch) {
    pool.submit(() -> handle(r));        // 处理是异步的
}
long next = batch.records(tp).get(batch.count() - 1).offset() + 1;
consumer.commitSync(Map.of(tp, new OffsetAndMetadata(next)));   // 提交却同步了
pool.shutdownNow();                       // 模拟崩溃：不等工作线程

[commit-ahead] polled=10, committed=10, actuallyProcessed=0
[commit-ahead] committed before crash={library.loan.created.async.51879-0=OffsetAndMetadata{offset=10, ...}}
[commit-ahead] consumer closed without waiting for workers (simulated crash)
[commit-ahead] restarted consumer firstOffset=10 (offsets 0..9 never processed)
[commit-ahead] lost=10

[callback] callbackThread=main exception=null
[callback] callback latency=52 ms (executed inside poll)

consumer.commitAsync(offsets, (o, e) -> {
    try {
        Thread.sleep(2000);        // 反例
    } catch (InterruptedException ie) {
        Thread.currentThread().interrupt();
    }
});

[reentry] callback thread=main enter
[reentry] callback slept 2000 ms inside poll
[reentry] this poll took 2024 ms

consumer.commitAsync(offsets, (o, e) -> {
    throw new IllegalStateException("boom from callback");
});

[callback] throwing callback: poll threw java.lang.IllegalStateException: boom from callback
[callback] throwing callback: exceptionSeen=true

[reentry] commitSync inside callback returned normally

private final AtomicReference<Map<TopicPartition, OffsetAndMetadata>> lastCommitted
        = new AtomicReference<>(Map.of());
private final AtomicReference<Map<TopicPartition, OffsetAndMetadata>> pendingRetry
        = new AtomicReference<>();
private final AtomicReference<Exception> fatal = new AtomicReference<>();

void commitAsyncSafely(KafkaConsumer<String, String> consumer,
                       Map<TopicPartition, OffsetAndMetadata> offsets) {
    consumer.commitAsync(offsets, (committed, e) -> {
        if (e == null) {
            lastCommitted.set(Map.copyOf(committed));
        } else if (e instanceof RetriableException) {
            // 可重试：记下来，下一轮 poll 后重提
            pendingRetry.set(Map.copyOf(committed));
        } else {
            // 不可重试（越权、被 fencing、组已失效）：停止消费，交给上层重启
            fatal.set(e);
        }
    });
}