《Java 消息中间件·Kafka 内核与客户端》9.2 拦截器链与埋点

拦截器在调用链上的位置

ProducerInterceptor 不是「发送完成后顺手调一下」的钩子。它在 KafkaProducer 里有明确的插入点，而且两个回调跑在不同线程上。

send() 的第一件事就是调 onSend()：

public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
    // intercept the record, which can be potentially modified; this method does not throw exceptions
    ProducerRecord<K, V> interceptedRecord = this.interceptors.onSend(record);
    return doSend(interceptedRecord, callback);
}

注意注释里的两句话：可以修改记录，本方法不抛异常。修改发生在 key/value 序列化和分区计算之前，所以拦截器改过的 key 会参与分区，改过的 value 会进入字节流。异常则被 ProducerInterceptors 捕获并记 WARN，不会传回业务代码。

onAcknowledgement() 的位置在 doSend() 内部的回调对象上：

@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
    if (metadata == null) {
        metadata = new RecordMetadata(topicPartition(), -1, -1, RecordBatch.NO_TIMESTAMP, -1, -1);
    }
    this.interceptors.onAcknowledgement(metadata, exception, headers);
    if (this.userCallback != null)
        this.userCallback.onCompletion(metadata, exception);
}

顺序很清楚：先拦截器，后用户回调。而且 AppendCallbacks.onCompletion 由 Sender 线程在收到响应（或最终失败）时调用，不是业务线程。

消费端同理。ConsumerInterceptor 的文档写得很直接：回调与调用 poll() 的线程相同；onConsume() 在 poll() 返回记录之前执行，可以过滤甚至生成新记录；onCommit() 在位移提交时执行。

一条消息会经过哪些回调

这张表有两个反直觉点：

1. 失败也可能走 onAcknowledgement，但线程不同。send() 内部抛 ApiException 时，doSend 的 catch 块调用 interceptors.onSendError(...)，最终还是进 onAcknowledgement，此时 metadata 是合成的：partition 可能已知，offset 一定是 -1。
2. onSend 看不到最终分区。如果 record 没指定 partition，分区在 doSend 里才计算，onSend 里看不到。想知道分区只能等 onAcknowledgement。

能改什么

onSend 可以改记录本身

onSend 返回的 ProducerRecord 就是后续流程用的记录。可以改 key、value、headers、topic、partition、timestamp。文档特别提醒：改 key/value 会影响分区计算，而且这种转换必须一致——同一个输入每次都要变成同一个输出，否则日志压缩会失效。

最常用的合法修改是加 header：

@Override
public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
    record.headers().add("x-send-ns",
            ByteBuffer.allocate(Long.BYTES).putLong(System.nanoTime()).array());
    return record;
}

onConsume 可以过滤和扩增

消费端拦截器可以返回任意数量的记录，包括 0 条或比输入更多。这很灵活，也有一个坑：过滤掉记录不会阻止位移前进。poll() 返回后，消费位置已经是这批记录的末尾；如果 onConsume 丢掉了中间几条，而业务随后提交位移，被丢的记录就再也不会被消费。

所以过滤逻辑要么同时记录被丢的 offset 并单独处理，要么放在业务层显式做，不要藏在拦截器里。

链式顺序与异常传播

配置多个拦截器时，按 interceptor.classes 里的顺序串行调用：

interceptor.classes=com.plume.mq.ChainA,com.plume.mq.ChainB,com.plume.mq.ChainC

第一个拿到客户端传入的 record，后面的拿到前一个的返回值。如果中间某个抛异常，异常被捕获记 WARN，下一个拦截器拿到的是上一个成功返回的 record，不是原始 record，也不是 null。

本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）：A 加 A、B 故意抛异常、C 加 C，消费端读到的 header 是：

链式 header x-chain=AC（B 抛异常，链继续）

日志里同时出现：

WARN org.apache.kafka.clients.producer.internals.ProducerInterceptors -
  Error executing interceptor onSend callback for topic: b4.intercept-lab, partition: null

这带来一个运维事实：拦截器坏了不会让发送失败，只会让埋点静默缺失。如果你的监控依赖拦截器上报，必须有独立的自检，不能只看「消息发出去了」。

不能改什么

ack 回调里的 metadata 是快照，headers 只读

RecordMetadata 描述的是已经发生的事实：topic、partition、offset、timestamp、序列化后大小。它没有 setter，改不了。

headers 在 ack 回调里也是只读的。本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）：在 onAcknowledgement 里执行 headers.add("x-ack-write", ...)，捕获到：

ack 回调里改 headers: IllegalStateException

原因是 doSend() 在追加批次前调用了 setReadOnly(record.headers())。onSend 阶段 headers 可写，onAcknowledgement 阶段只读，这个分界由源码保证。

onAcknowledgement 在 Sender 线程，不能阻塞

文档明确写了：这个方法一般在后台 I/O 线程执行，实现必须足够快，否则会延迟其他线程的消息发送。

Sender 是单线程。一个分区的一批消息确认后，onAcknowledgement 逐条执行；如果每条睡 1 ms，一批 500 条就占住 Sender 半秒，期间所有分区的请求都发不出去。这里适合做计数、打点、写无锁队列，不适合做远程调用、查库、打印大日志。

onSend 在调用线程，阻塞会等量拖慢 send()

onSend 不是「后台预处理」，它就在业务线程里。本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29），100 次 send()+flush()：

100 次 send+flush 总耗时: 无拦截器=17ms, onSend sleep(1)=157ms
（单次运行，随调度波动，不作基准）

差值 140 ms，和 100 次 sleep(1) 的量级一致（macOS 上 Thread.sleep(1) 通常略多于 1 ms）。结论很直接：onSend 里做的任何事，都会出现在业务接口的 P99 里。

不要返回 null，也不要跨回调持有 record 引用

onSend 返回 null 会让 doSend(null) 在取 topic 时失败。拦截器应该「原样返回」或「返回一个新 record」，不要用 null 表示「不处理」。

另一个常见错误是在 onSend 里把 record 塞进一个 Map，指望在 onAcknowledgement 里按对象身份取回来。AppendCallbacks 只保留 topic、partition 和 headers，不保留 record 引用；重试时 Sender 重发的是批次字节，也不会把原 record 带回来。跨回调关联要靠 header 或业务主键。

埋点位置怎么选

发送耗时：用 header 把 onSend 和 ack 连起来

onAcknowledgement 拿不到「这条记录是什么时候 send 的」。正确做法是在 onSend 写一个时间戳 header，在 ack 里读出来：

public final class TimingInterceptor implements ProducerInterceptor<String, String> {
    static final ConcurrentLinkedQueue<Long> LATENCIES = new ConcurrentLinkedQueue<>();

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        record.headers().add("x-send-ns",
                ByteBuffer.allocate(Long.BYTES).putLong(System.nanoTime()).array());
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception, Headers headers) {
        var h = headers.lastHeader("x-send-ns");
        if (h != null) {
            long sentNs = ByteBuffer.wrap(h.value()).getLong();
            LATENCIES.add(System.nanoTime() - sentNs);
        }
    }

    @Override
    public void configure(Map<String, ?> configs) { }

    @Override
    public void close() { }
}

4.3.0 的 ProducerInterceptor 有三参数 onAcknowledgement(metadata, exception, headers)，默认实现委托给两参数版本。要用 header 关联，就必须覆写三参数版本；只覆写两参数版本拿不到 headers。

为什么不用 Map<String, Long> 按 key 关联？因为 key 会重复，同一 key 的多条消息会互相覆盖。header 跟着记录走，重试和批处理都不会丢。

实测 100 条

配置 linger.ms=10、acks=all，先发一条 warmup 让元数据就绪，再发 100 条。本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）：

acks=100 errors=0 samples=100
send→ack 微秒: min=5429 p50=6958 p95=8245 max=8440 mean=6926.9
同一次运行 request-latency-avg=8.0ms record-queue-time-avg=10.5ms batch-size-avg=1893.0

三个数字的口径完全不同：

send→ack 包含排队、linger、批次组装、网络、broker 处理和回包，不是纯网络 RTT。批次里靠后的记录排队时间短，所以单条均值会低于批次的 record-queue-time-avg。要做 SLA，用 send→ack；要定位是哪一段慢，再拆到 record-queue-time-avg 和 request-latency-avg。

消费端埋点

ConsumerInterceptor.onConsume() 在记录已经拉取、已经反序列化之后执行，所以它测不到网络和 broker 的耗时。它适合做：按业务维度计数、过滤、补充上下文、记录「进入业务逻辑」的时间。

要测拉取耗时，用消费者自带的 fetch-latency-avg；要测端到端，把生产端的 x-send-ns header 透传到业务处理完，再和当前时间相减。注意这个数字包含消息在 topic 里停留的时间，不是「消费有多慢」。

onCommit() 只在提交成功时调用。本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）：

ConsumerInterceptor onConsume 记录数=302 onCommit 次数=1

302 条记录、一次同步提交，所以 onCommit 只触发一次。它适合记录「提交成功了几次、提交了哪些 offset」，不适合统计消息条数——条数在 onConsume 里。

拦截器自己也能注册指标

4.3.0 的 ProducerInterceptor / ConsumerInterceptor 文档里有一句容易漏掉的话：实现 org.apache.kafka.common.metrics.Monitorable 可以让拦截器注册自己的指标，注册时会自动加上两个 tag：config=interceptor.classes 和 class=<拦截器类名>。

这比在拦截器里维护静态 AtomicLong 更适合生产：指标会出现在客户端的 metrics() 里，和 record-send-total 这些内置指标同一套采集路径。静态计数器的缺点是进程内可见、跨实例无法区分，而且测试代码容易互相污染。

如果不实现 Monitorable，退而求其次是把计数器暴露成只读方法，由应用自己的采集线程定期读取。无论哪种，都要保证读操作不加锁、不阻塞 Sender。

生命周期

拦截器的 configure() 在客户端构造时调用，close() 在客户端关闭时调用。ProducerInterceptors.close() 会逐个关闭，单个拦截器关闭抛异常只记 ERROR，不影响其他拦截器。

有两个边界要记住：

• configure() 拿到的配置包含 producer 的全部配置，多个拦截器共享同一个配置命名空间，自定义 key 要加前缀，避免和 bootstrap.servers、acks 这类名字冲突。
• close() 不保证在进程被 kill 时执行。依赖 close() 落盘的统计会丢最后一段数据，重要指标应该持续上报，而不是攒到关闭时一次写。

常见错误清单

还有一条线程安全要求：ProducerInterceptor 的回调可能来自多个线程（多个业务线程调 send()），实现里的共享状态必须用并发容器或原子类。ConsumerInterceptor 的回调固定在 poll() 线程，但不要在实例里缓存与某次订阅绑定的状态，换一个消费者或重新订阅后它就不成立了。

本节判据

读完这一节，你应该能回答三个问题：

1. onSend 和 onAcknowledgement 分别在哪跑？ 前者在调用 send() 的业务线程，后者在 Sender 线程。
2. 拦截器抛异常会怎样？ 被捕获并记 WARN，链继续，发送不受影响；代价是埋点静默缺失。
3. 测发送耗时为什么不能只看 request-latency-avg？ 它只覆盖一次请求，不含排队和 linger；单条记录的完整体验要用 header 在 onSend 与 ack 之间关联。

下一节把这些零散的打点收拢成一套指标口径：metrics() 返回的 77 项到底是什么，哪些能反映背压，哪些只是结果而不是原因。

阅读导航：上一节 9.1 序列化器与 Schema 边界 ｜ 下一节 9.3 客户端指标口径