本文围绕 Kafka 的核心运行机制展开，内容包括日志保留与清理、生产者内部结构、分区与顺序性、批处理与重试、可靠性配置、消费者轮询、消费者组重平衡、位移提交，以及异步消费的设计取舍。
文中配置项的默认值可能随 Kafka 版本变化。实际使用时，应以目标版本的官方配置文档为准。
Kafka 的删除单位是日志段（Log Segment），而不是单条消息。
每个分区（Partition）的日志会被拆分为多个日志段文件。使用默认删除策略时，Kafka 不会逐条删除过期消息；当某个日志段满足清理条件后，Kafka 会删除整个日志段。
当前正在写入的日志段称为活动日志段（Active Segment）。无论其中的消息有多旧，活动日志段都不会被删除。只有当它滚动为非活动日志段后，才可能参与清理。
日志段在满足下列任一条件时发生滚动（Roll）：
segment.bytes，文中默认值为 1 GB；segment.ms，文中默认值为 7 天。后台线程按照 log.retention.check.interval.ms 周期检查可清理的日志段，文中默认检查间隔为 5 分钟。
这意味着，即使某条消息已经超过保留时间，也不一定会被立即删除。它所在的日志段必须先关闭，并在后续检查中整体满足删除条件。
cleanup.policy：选择日志清理策略cleanup.policy 决定主题使用哪种清理方式。
delete默认策略。根据保留时间或保留大小删除旧日志段，适合事件流、日志、指标等按时间推进的数据。
compact日志压缩策略（Log Compaction）。对于相同的 Key，Kafka 最终只保留最新值；旧值会在后台清理。Value 为 null 的消息称为墓碑消息（Tombstone），用于表示删除某个 Key。
该策略适合保存“当前状态”的主题，例如每个用户的最新资料。Kafka 内部主题 __consumer_offsets 就使用日志压缩。
compact,delete同时启用日志压缩和按时间、大小删除。
delete 策略的两个触发条件在 cleanup.policy=delete 时，下面两个条件谁先达到，就由谁触发清理：
retention.ms，文中默认值为 604800000，即 7 天；设置为 -1 表示永久保留。retention.bytes，文中默认值为 -1，表示不限制大小。retention.bytes 是每个分区的限制，不是整个主题的总限制。假设一个主题有 10 个分区，每个分区的 retention.bytes 都是 10 GB，那么主题的理论日志规模可以接近 100 GB，副本占用尚未计算在内。
本文所说的 Kafka 客户端或 Producer 进程，是一个宏观容器。实例化一个 KafkaProducer 后，其内部主要包含三个组件：
NetworkClient 与 Broker 进行网络通信。可以把一条消息的发送过程概括为：
业务代码调用 send()
→ ProducerInterceptor.onSend()
→ Key/Value 序列化
→ 选择 Partition
→ 写入 RecordAccumulator
→ Sender 组成 ProduceRequest
→ Broker 写入并返回响应
→ ProducerInterceptor.onAcknowledgement() / CallbackProducerInterceptor 有两个核心方法：
onSend()：业务线程调用 send() 后，在消息进入后续发送流程时执行；onAcknowledgement()：Sender 收到服务端确认或发送失败时执行。Producer 默认不执行自定义拦截逻辑。可以通过 interceptor.classes 配置一个或多个拦截器。多个拦截器按顺序执行，形成责任链（Chain of Responsibility）。
拦截器应保持轻量，尤其不要在确认回调路径中执行耗时的数据库或网络操作，否则会阻塞 Sender 线程。
Producer 发送消息前，需要知道：
当 Producer 第一次遇到某个 Topic，或本地元数据已过期时，主线程会请求 Sender 更新元数据，并等待结果。由于 Topic 元数据通常不会频繁变化，Producer 会把结果缓存在本地，避免每次发送都请求集群。
metadata.max.age.ms 控制元数据缓存的最大存续时间。即使没有发生显式错误，达到该时间后，Producer 也会主动刷新元数据。
元数据刷新涉及两类退避（Backoff）：
请求超时、连接断开，或返回可重试错误时，说明本次请求真正失败。客户端不会立即高频重试，而会等待一段时间。
该机制不是简单比较两次响应的字节是否相同，而是基于 Topic-Partition 的 Epoch 判断进展：
等价响应退避只发生在客户端明确等待 Epoch 推进的按需刷新过程中，例如收到 NotLeaderOrFollowerException、等待新 Leader 时；普通的周期性刷新不会因此进入等价响应退避。相关控制标记为 backoffOnEquivalentResponses。
海量消息不会进入一个全局队列，而是被路由到 Topic 的不同 Partition。不同 Partition 往往分布在不同 Broker 上。
分区决策可以概括为：
Partitioner），执行自定义策略；null，使用粘性分区策略（Sticky Partitioner），以批次为粒度在分区之间切换。默认分区逻辑可简化为：
partition = toPositive(murmur2(serializedKeyBytes)) % numPartitions这条公式带来四个重要结论。
只要序列化后的 Key 字节和分区数量不变，相同 Key 就会映射到相同 Partition。这是 Kafka 实现同 Key 分区内有序的基础。
Kafka 哈希的是 Key 序列化后的字节，而不是 Java 对象本身。因此，序列化器必须稳定。
如果同一个业务 Key 有时使用不同字符编码、包含额外空格，或者 JSON 字段顺序发生变化，最终字节就可能不同，消息会被路由到不同 Partition，顺序也会被拆散。
分区结果依赖 numPartitions。增加 Partition 后，旧消息不会移动，但同一个 Key 的新消息可能映射到另一个 Partition。因此，扩容分区可能破坏跨扩容时间点的 Key 顺序。
Key 为 null 时会跳过哈希逻辑，使用粘性分区器。粘性分区不是逐条消息做简单轮询，而是尽量让一批消息先进入同一 Partition，以提高批处理和压缩效率。
多个 Producer 向同一个 Topic 发送相同 Key 很常见。只要它们使用相同的序列化结果、分区器和分区数量，这些消息仍会进入同一 Partition，Kafka 会在该 Partition 的日志中形成一个确定的物理顺序。
但 Kafka 无法推断不同 Producer 之间的业务因果顺序。例如，“创建订单”和“支付成功”由两个独立 Producer 并发发送，即使最终进入同一 Partition，谁先到达 Broker 仍取决于调用时机和网络时序。Kafka 能保证写入日志后的分区内顺序，但不能替业务系统决定哪一个事件本应先发生。
因此，涉及状态流转时，消费端仍需要状态机、幂等和重试机制：
void consumePaymentMessage(PaymentMsg msg) {
String orderId = msg.getOrderId();
// 在数据库事务中通过行锁保护订单状态。
// SELECT * FROM orders WHERE id = ? FOR UPDATE
Order order = db.queryOrderForUpdate(orderId);
if (order == null) {
// 支付事件先到，但订单尚未创建：拒绝本次处理并稍后重试。
throw new OrderNotFoundException(
"订单尚未创建，支付消息暂不处理"
);
}
if (order.getStatus() == OrderStatus.INIT) {
db.updateStatus(orderId, OrderStatus.PAID);
} else if (order.getStatus() == OrderStatus.PAID) {
// 幂等：已支付时不重复更新。
log.info("Order {} is already paid", orderId);
}
}示例中的字符串状态比较已改为枚举，以避免 Java 使用
==比较字符串所带来的错误。
RecordAccumulator 负责暂存尚未发送的消息，其中两个核心部分是：
free：空闲内存池，类型为 BufferPool；总容量由 buffer.memory 控制。ProducerBatch。从逻辑上可以理解为：
Topic
├─ Partition 0 → deque<ProducerBatch>
├─ Partition 1 → deque<ProducerBatch>
└─ Partition 2 → deque<ProducerBatch>Producer 在 Partition 维度积攒批次。ProducerBatch 中的数据最终写入从 BufferPool 申请的 ByteBuffer。
Kafka Producer 对整个消息批次进行压缩，而不是逐条压缩。压缩算法由 compression.type 配置，可选值包括：
nonegzipsnappylz4zstd批次越充实，通常压缩率越好，因此批处理与压缩应结合调优。
batch.size：大小触发器batch.size 表示单个 Partition 批次的目标上限，文中默认值为 16 KB。
常见误解包括：
当单条消息大于 batch.size 时，Producer 会为它分配更大的批次。batch.size 不是单条消息的大小上限；相关上限由 max.request.size 等配置约束。
增大 batch.size 可以减少请求数量并改善压缩率，但会提高内存占用，也可能增加等待批次填充的时间。
linger.ms：时间触发器linger.ms 表示批次尚未达到 batch.size 时，Producer 愿意主动等待更多消息的时间。
批次在以下任一条件满足时具备发送资格：
达到 batch.size，或等待达到 linger.ms；谁先发生，以谁为准。linger.ms=0 不等于“完全没有批处理”，而是“不额外引入等待”。在高负载下，一个请求尚在发送时，新消息仍会自然积累为下一批。
适当提高 linger.ms，例如 5～50 ms，可以用少量单条延迟换取更大的批次、更高的吞吐和更好的压缩效果。
buffer.memory：总内存池与背压上限buffer.memory 是 Producer 用于暂存待发送消息的总内存预算，文中默认值为 32 MB。它作用于整个累加器，而不是单个批次。
当业务生产速度超过 Sender 的发送速度时，内存池会逐渐耗尽。后续 send() 将在业务调用线程中阻塞，最多等待 max.block.ms，随后抛出 TimeoutException。
这也是 Producer 的背压阀门：send() 虽然主要是异步接口，但并不保证调用线程永远立即返回。
batch.size 决定批次的大小触发条件；linger.ms 决定批次的时间触发条件；buffer.memory 决定全部待发送批次可占用的总内存，以及何时对业务线程施加背压。高吞吐场景通常会适当增加 linger.ms、batch.size 和 buffer.memory，并启用 lz4 或 zstd 压缩。低延迟场景则应保持较小的 linger.ms 和适中的批次。
调大 batch.size 时，需要结合活跃 Partition 数量评估内存。多个活跃 Partition 可能各自持有一个或多个批次，因此内存需求会随活跃分区数量增长。不过，实际占用并不总是严格等于 batch.size × 活跃分区数，因为批次按需分配，且还受消息大小、并发请求和释放速度影响。
max.block.ms：业务线程同步阻塞上限max.block.ms 控制业务线程在 KafkaProducer.send() 前置阶段等待资源的最长时间，文中默认值为 60 秒。
它主要覆盖：
BufferPool 中等待可用内存。这两个阶段的等待时间累计计算。超过 max.block.ms 后，send() 会在调用线程同步抛出 TimeoutException。
max.block.ms 不用于等待 Broker ACK，它与后台网络确认阶段属于不同范围。
request.timeout.ms：单次网络请求超时request.timeout.ms 控制 Sender 发出一次 ProduceRequest 后，等待 Broker 响应的最长时间，文中默认值为 30 秒。
如果超过该时间仍未收到响应，本次网络请求被判定为失败。只要消息尚未超过总投递时限，并且错误允许重试，Sender 就可以进入退避和重试流程。
retry.backoff.ms：重试退避时间retry.backoff.ms 是可重试失败后再次尝试前的基础等待时间，文中默认值为 100 ms。连续失败时，退避时间可以逐步增长，直至 retry.backoff.max.ms 所限定的上限。
该等待发生在 Sender 的调度逻辑中，不会直接休眠或占用业务主线程。其作用是避免 Leader 选举、Broker 重启、长时间 GC 等故障期间产生高频无效重试和网络风暴。
如果 Broker 很快明确返回可重试错误，例如 NotLeaderOrFollowerException，当前请求会立即结束，无须继续等满 request.timeout.ms，随后进入退避。
如果返回不可恢复错误，例如鉴权失败或消息过大，Sender 会立即结束该消息，不再等待退避或总投递时限。
delivery.timeout.ms：端到端投递上限delivery.timeout.ms 限制消息从进入 Producer 累加器，到最终成功或失败的总时长，文中默认值为 120 秒。
它覆盖：
RecordAccumulator 中等待组成批次的时间；超过该时限后，Producer 会使相关消息失败，完成对应的 Future，并通过 Callback 报告 TimeoutException。该机制防止 Broker 长期不可达时无限重试并永久占用缓冲内存。
配置时应满足：
delivery.timeout.ms >= request.timeout.ms + linger.ms阶段 A：send() 前置准备
受 max.block.ms 约束
├─ 等待 Metadata
└─ 等待 BufferPool 内存
阶段 B：消息进入 RecordAccumulator
delivery.timeout.ms 开始约束后续生命周期
├─ 等待批次发送
├─ 单次请求受 request.timeout.ms 约束
├─ 失败后等待 retry.backoff.ms
├─ 再次请求
└─ 直至成功、不可重试失败或达到 delivery.timeout.ms需要注意：实现细节和计时边界可能随 Kafka 客户端版本变化。理解参数的职责边界，比把某一源码版本中的纳秒级起点当作固定契约更重要。
当客户端已经把某次请求判定为超时后，迟到的响应通常不会使已安排的重试自动撤销。网络请求通过关联编号（Correlation ID）匹配响应；请求脱离待响应状态后，迟到响应无法再作为该次尝试的正常完成结果。
这也是网络超时场景可能产生重复写入的原因之一：客户端无法确定 Broker 是否已经成功写入，只能重试。启用幂等 Producer 后，Broker 可以利用 Producer ID 和序列号识别并抑制这类重复。
acks、副本与可靠性acks=0：发送后不等待确认Producer 将记录写入网络后即认为发送完成，不等待 Broker 响应。
优点是延迟最低；代价是没有投递确认。如果 Leader 不可用、网络丢包或消息丢失，Producer 可能完全不知道。由于没有 ACK，也缺少基于服务端响应触发重试的依据。
仅适合允许少量丢失的数据，例如部分遥测、调试日志或非关键指标。
acks=1：等待 Leader 确认Leader 把消息追加到本地日志后即可返回成功，不等待 Follower 完成复制。
风险窗口是：Leader 返回成功后、Follower 复制前发生故障；若随后由尚未拥有该消息的 Follower 当选为新 Leader，这条消息可能丢失。
它以较弱的故障转移安全性换取较低确认延迟。
acks=all / acks=-1：等待 ISR 确认Leader 需要等待当前同步副本集合（In-Sync Replicas，ISR）达到写入确认条件后才返回成功。这是最强的 ACK 模式。
代价主要是确认延迟增加，因为需要等待副本复制。不过 Producer 可以流水线发送其他批次，因此吞吐下降通常没有单条延迟增长那么直接。
acks=all 本身仍不够“等待所有 ISR”有多安全，取决于 ISR 中实际还有多少个副本。如果 ISR 已缩减到只剩 Leader，那么 acks=all 在副本数量层面可能接近 acks=1。
防线是 Broker 或 Topic 级配置 min.insync.replicas。
生产环境常见组合是：
replication.factor = 3
min.insync.replicas = 2
acks = all当 ISR 少于 2 时，Broker 会拒绝写入并返回副本不足错误。这个组合选择牺牲部分可用性，以避免只剩一份数据时继续接受关键写入。
min.insync.replicas 只有在 Producer 使用 acks=all 时才能提供这种约束；使用 acks=1 时，Leader 不需要等待足够的 ISR 确认。
acks=all 不等于立即 fsyncKafka 的持久性主要来自多个 Broker 上的副本，而不是每条消息都同步刷盘。消息可能位于多台机器的页缓存（Page Cache）中；跨机器复制降低了单机故障导致数据丢失的风险。
acks=all幂等 Producer 通过 Producer ID（PID）和序列号（Sequence Number）让 Broker 识别重试造成的重复批次，并维护分区内顺序。相关状态必须与分区日志一起在副本之间保持一致。
禁用幂等性会产生三类隐性代价：
acks=1 时存在 Leader 故障转移丢失窗口。启用幂等性后，Kafka 可以利用序列号约束重复和分区内重试顺序。显式配置与幂等性要求冲突的 acks 值时，客户端会拒绝不一致配置或需要禁用幂等性，具体行为取决于客户端版本与其他配置。
acks 不直接决定消费者可见性acks 主要决定 Producer 在什么条件下认为写入成功。消费者通常只能读取到高水位（High Watermark）以内的数据，因此消费者可见范围仍受副本提交进度约束。
producer.send(record);调用方丢弃返回的 Future。吞吐很高，但如果最终发送失败，应用可能无法感知。只适合确实允许偶发丢失的数据。
producer.send(record).get();业务线程等待 Broker 确认或异常。它便于逐条处理错误，但会破坏流水线并显著降低吞吐。
get() 可能抛出 InterruptedException，并使用 ExecutionException 包装真实原因，需要通过 getCause() 查看底层异常。
原生客户端：
producer.send(record, (metadata, exception) -> {
if (exception != null) {
// 记录失败指标和必要上下文。
}
});Spring Kafka 可通过返回的异步结果注册完成处理：
kafkaTemplate.send(topic, key, value)
.whenComplete((result, exception) -> {
if (exception != null) {
// 记录失败指标和必要上下文。
}
});异步发送既保留批处理和流水线吞吐，也能得到最终结果，适合大多数应用。
即使使用异步发送，仍有部分异常会直接从调用线程的 send() 抛出，例如序列化失败、获取元数据超时或缓冲池耗尽。
Broker 拒绝、重试耗尽和投递超时等错误，则通常通过 Future 或 Callback 异步返回。
因此，健壮代码需要两条异常处理路径：
try {
producer.send(record, (metadata, exception) -> {
if (exception != null) {
metrics.increment("kafka.send.failed");
log.warn(
"Kafka send failed, topic={}, key={}, traceId={}",
record.topic(), record.key(), traceId, exception
);
}
});
} catch (RuntimeException exception) {
// 处理 send() 阶段的同步失败。
metrics.increment("kafka.send.rejected");
throw exception;
}Producer Callback 通常在 Sender I/O 线程中执行。如果在 Callback 内进行数据库写入、HTTP 调用或阻塞等待，会拖慢所有 Partition 的发送。
Callback 中适合做：
日志应包含 Topic、Partition、业务 Key、Trace ID 等上下文，但不应默认打印完整消息体，以免产生敏感信息泄漏和日志膨胀。
对于不能丢失的核心数据，应设计持久化兜底，例如事务发件箱、失败消息表或死信主题。不能只依赖 Callback 临时再发一次，因为应用崩溃时内存中的补偿任务也会丢失。
在启用正确的顺序和幂等配置时，同一 Partition 的 Callback 会按照记录顺序执行；但仍不应让 Callback 承担复杂业务处理。
poll() 模型poll() 是消费者的核心驱动器poll() 驱动 Kafka Consumer 的主要工作，包括：
KafkaConsumer 被设计为非线程安全。通常应由一个线程独占，遵循：
poll → process → poll → process不要让多个业务线程同时调用同一个 KafkaConsumer 实例。
poll(timeout) 是最大阻塞时间，不是轮询周期consumer.poll(Duration.ofMillis(100));它表示：当本地没有可立即返回的数据时，最多等待 100 ms；如果数据提前到达，就立即返回；到时仍无数据，则返回空集合。
它不表示“每 100 ms 自动执行一次 poll”。实际调用频率由应用循环和每批消息的处理时间共同决定。
max.poll.interval.ms：逻辑存活检查max.poll.interval.ms 控制两次 poll() 调用之间允许的最大时间，文中默认值为 5 分钟。
如果业务处理太慢，长时间没有再次调用 poll()，协调器会认为该消费者虽然进程可能还活着，但已经无法正常推进消费，于是将其移出消费者组并触发重平衡。
Kafka 需要判断两个不同问题：
物理存活：进程是否还活着？
heartbeat.interval.ms 发送；session.timeout.ms 未收到心跳，协调器认为 Consumer 已失联。逻辑存活：业务是否仍在推进？
max.poll.interval.ms 检测；process() 中，也会被视为处理失效。新消费者组协议对心跳参数的控制位置有所变化，但“物理存活”和“处理进度”是两个不同问题这一点不变。
fetch.min.bytes 与 fetch.max.wait.ms 共同控制 Broker 如何响应拉取请求：
fetch.min.bytes：Broker 希望至少积攒多少数据再返回；fetch.max.wait.ms：即使未达到最小数据量，最多等待多久也必须返回。它们与 Producer 的 batch.size、linger.ms 在概念上近似镜像：前者控制 Broker 向 Consumer 返回数据的批量程度，后者控制 Producer 向 Broker 发送数据的批量程度。
但两组参数所处位置和实现不同，不能把它们视为完全相同的机制。
max.poll.records 只限制返回条数max.poll.records 限制一次 poll() 最多交给业务代码多少条记录，并不直接限制网络层一次从 Broker 拉取多少数据。
Consumer 可能已经通过网络获取较大的一批数据，并缓存在内部队列中。假设本地缓存有 10,000 条，而 max.poll.records=500：
poll() 最多返回前 500 条；poll() 可以继续从本地缓存返回，而不必先发起新的网络请求。网络拉取规模还受 fetch.max.bytes、max.partition.fetch.bytes 等参数影响。
Kafka 使用内部主题 __consumer_offsets 保存消费者组元数据和已提交位移。某个消费者组会根据 group.id 映射到 __consumer_offsets 的一个分区，该分区 Leader 所在 Broker 上的 GroupCoordinator 负责协调这个消费者组。
Consumer 启动后，可以向任意 Broker 发送 FindCoordinatorRequest。Broker 返回目标 GroupCoordinator 的地址，后续消费者组协议在 Consumer 与该协调器之间进行。
经典协议主要包含以下过程：
JoinGroupRequest；SyncGroupRequest 把方案交给协调器；SyncGroupResponse 把各自结果下发给所有成员。把分配算法放在客户端，使用户能够实现自定义 PartitionAssignor，无需升级 Broker。但它也让重平衡逻辑、监控和版本兼容更加复杂。
早期重平衡要求成员先撤销全部 Partition，再等待新分配方案。在此期间，整个消费者组暂停处理，这种现象称为 Stop-the-World。
同步屏障（Synchronization Barrier）确保在重新分配期间，一个 Partition 不会同时被两个 Consumer 正常拥有，但代价是即使归属最终不变的 Partition 也要暂停。
KIP-429 引入增量协作式重平衡（Incremental Cooperative Rebalancing），目标是只撤销归属确实发生变化的 Partition。
其思路是：
该机制改善了 Stop-the-World，但经典协议下的部分组操作在重平衡期间仍受影响，客户端也仍承担较多协调逻辑。
KIP-848 引入下一代消费者重平衡协议。原文描述的 Kafka 4.0 行为是：
group.protocol=consumer 选择新协议；ConsumerGroupHeartbeatRequest 与 ConsumerGroupHeartbeatResponse 完成成员元数据上报和分配协调；JoinGroup、SyncGroup、LeaveGroup 屏障流程；partition.assignment.strategy 不再承担同样角色。Consumer 收到新的分配结果后，会比较当前拥有的 Partition 与目标集合：撤销不再属于自己的 Partition，添加新获得的 Partition。服务端统一掌握分配状态，有助于减少客户端协调复杂度和不必要的全组停顿。
常见触发条件包括：
max.poll.interval.ms；重平衡发生在同一个消费者组内部，不同 group.id 的消费者组互不参与对方的分区分配。
经典协议通过 partition.assignment.strategy 配置分配策略。
RangeAssignorRange 策略尊重 Topic 边界：对每个 Topic 分别按 Partition 编号切分，再分配给 Consumer。
如果不能整除，靠前的 Consumer 会多承担一个 Partition。订阅多个且分区数量相同的 Topic 时，同一 Consumer 往往能获得这些 Topic 中相同编号范围的 Partition，这有利于协同分区（Co-partitioning）和部分流式 Join 场景。
缺点是容易让靠前的 Consumer 在多个 Topic 上重复多拿 Partition，造成负载倾斜。
RoundRobinAssignorRound Robin 会把所有订阅 Topic 的 Partition 放入统一序列，再轮流分配给 Consumer。它打破 Topic 边界，通常能获得更均匀的 Partition 数量。
当成员数或 Partition 数变化时，较多 Partition 的归属可能改变，因此不适合分区迁移成本很高的有状态消费者。
StickyAssignorSticky 策略在尽量均衡的同时，优先减少 Partition 归属变化，适合重分配成本较高的场景。
假设存在：
Topic A: A-P0, A-P1
Topic B: B-P0, B-P1如果希望不同 Topic 的相同编号 Partition 稳定对齐，通常需要它们具有一致的 Partition 数，并使用合适的分区键和分配策略。仅靠 Range 分配并不能自动保证任意流式 Join 的正确性；生产端分区规则也必须兼容。
也称 Position 或 Fetch Position，表示下一次 poll() 将交付的记录位移。它保存在 Consumer 本地内存中，并随着 poll() 返回记录而推进。
例如，poll() 返回到 Offset 6 后，Position 变为 7。
表示消费者组持久化到 Kafka 的恢复点，按以下组合独立保存：
(group, topic, partition)例如：
| Group | Topic | Partition | Committed Offset |
|---|---|---|---|
order-group | order | 0 | 1050 |
order-group | order | 1 | 2034 |
order-group | order | 2 | 899 |
提交值是“下一条希望读取的 Offset”，即：
committed offset = last processed offset + 1提交 7 表示 Offset 0～6 已处理完成，下次从 7 恢复。这里出现差一错误（Off-by-One）可能导致重复读取或跳过消息。
Position 会在拉取时向前推进；Committed Offset 只会在自动提交或显式提交时推进。
两者之间的区间表示“已经交给应用，但尚未被安全提交”的消息。如果 Consumer 此时崩溃，新实例会从 Committed Offset 恢复，这部分消息将再次被消费。
这正是至少一次处理语义中重复消息的来源之一。
常见消费延迟定义为：
lag = log-end offset - committed offset其中：
由于 Lag 通常基于 Committed Offset，它反映的是持久化进度，不一定等于 Consumer 内存中的实际处理进度。
启用 enable.auto.commit=true 后，Consumer 会在 poll() 过程中检查自动提交时间。达到 auto.commit.interval.ms 后，提交此前 poll() 推进的当前位置。
自动提交隐含一个假设：上一次 poll() 返回的消息已经处理完毕。
同步处理时，只要在下一次自动提交发生前完成上一批消息，风险相对可控；但如果把消息交给后台线程池，主线程继续快速 poll()，就可能在后台任务完成前提交较大的 Offset。
例如：
poll() 返回 Offset 0～499；poll()，触发自动提交 500；自动提交也可能造成重复：如果消息已经处理完，但在下次自动提交前进程崩溃，恢复后会再次处理这些消息。
使用：
enable.auto.commit=false核心原则是：处理完成后再提交。如果处理成功后、提交前崩溃，消息会重放，从而形成至少一次（At-Least-Once）语义。
commitSync()commitAsync()常见组合是：循环中使用 commitAsync()，关闭或 onPartitionsRevoked() 中使用 commitSync() 做最后确认。
auto.offset.resetauto.offset.reset 按单个 Partition 判断，主要在两种情况下触发：
group.id 或刚增加的 Partition；假设 Topic 新增 P3，而消费者组对 P0、P1、P2 已有提交记录。只有 P3 会触发 auto.offset.reset，原有 Partition 继续从自己的 Committed Offset 消费。
earliest从当前仍被保留的最旧位置，即 Log Start Offset 开始。这里的“最早”不一定是 0。
latest从 Log End Offset 开始，只消费随后到达的新消息。现有但尚未消费的数据会被跳过，因此对不能丢数据的业务要谨慎使用。
none不自动重置。如果没有有效提交位移，则把错误暴露给应用，由业务自行决定如何处理。
假设 Committed Offset 为 100，但 Offset 100～200 已被删除，当前 Log Start Offset 为 201：
auto.offset.reset 选择新位置；earliest 会通过查询日志边界把位置重置到 201；latest 会跳到当前 Log End Offset；none 则向应用报告错误。已经被日志清理的数据无法通过重置找回。
__consumer_offsets 自身的保留消费者组的提交位移保存在 __consumer_offsets 中，也受位移保留策略管理。长期不活跃的消费者组，其位移可能被清理；再次启动时，由于没有有效 Committed Offset，会触发 auto.offset.reset。
原文给出的旧版本默认值是 offsets.retention.minutes=10080，即 7 天。该默认值和过期判定规则可能随 Kafka 版本变化，应按部署版本确认。
seek()seek(TopicPartition, offset)：跳到精确 Offset；下一次 poll() 从该位置读取。seekToBeginning(partitions)：跳到当前仍保留的最旧 Offset。seekToEnd(partitions)：跳到日志末尾，只读取后续新消息。offsetsForTimes(...)：查找时间戳不小于目标时间的第一条记录位置，可用于“重放下午 2 点后的消息”。position(tp)：查询当前内存位置。committed(...)：查询已提交位移。beginningOffsets(...) / endOffsets(...)：只查询日志边界，不移动位置。使用 subscribe() 时，Consumer 要等重平衡完成后才真正获得 Partition，而重平衡通常由 poll() 驱动。因此，刚 subscribe() 就调用 seek()，可能出现：
IllegalStateException: No current assignment for partition正确位置包括：
ConsumerRebalanceListener.onPartitionsAssigned()；poll() 完成并获得分配之后。如果使用 assign(partitions) 手动指定 Partition，则不参与消费者组自动分配，可以立即 seek()。批处理重放工具常使用这种方式获得确定性控制。
seek() 只改变 Consumer 内存中的 Position，不会自动提交。
seek() 后重读，但不提交回退位置；重启或重平衡后仍从原 Committed Offset 恢复。seek() 后提交新位置；新的起点在重启后仍然有效。seek() 的延迟生效特性seek() 只设置位置，真正拉取发生在下一次 poll()。如果在两次 poll() 之间对同一 Partition 连续调用多次 seek()，最后一次设置生效。
目标位置低于 Log Start Offset 或高于 Log End Offset 时，后续行为与客户端版本及 auto.offset.reset 配置有关。生产代码不应依赖模糊的越界回退，应先通过 beginningOffsets() 和 endOffsets() 校验目标位置。
创建新的 group.id，配合 auto.offset.reset=earliest 或手动 seek() 读取历史数据。新消费者组拥有独立位移，不会改变线上主消费者组的进度。
使用 kafka-consumer-groups --reset-offsets 或 AdminClient 修改现有消费者组位移。这会永久改变主业务的消费起点，通常应先停止相关消费者，并严格核对 Topic、Partition、目标 Offset 和执行范围。
最容易保证正确性的模型是：
一个 KafkaConsumer 由一个线程独占
→ poll()
→ 在该线程中同步处理
→ 处理完成后提交需要并行时，优先增加 Partition，并增加 Consumer 实例或每个实例中的独立消费线程；每个线程仍独占一个 Consumer，并同步处理自己获得的 Partition。
这种方式让 Partition 顺序、Offset 提交、重平衡和优雅关闭保持简单。
同一 Partition 的消息按顺序交给线程池后，任务可能乱序完成。如果直接提交“已完成任务中的最大 Offset”，较小 Offset 对应的任务可能尚未完成；进程崩溃后，这些任务会被永久跳过。
正确做法是按 Partition 跟踪全部在途 Offset，只提交连续完成的最大前缀（Contiguous Completed Prefix），也称水位线提交（Watermark Commit）。
主线程持续 poll()，后台线程池却处理不过来，会导致本地队列无限增长，最终耗尽内存。反过来，如果主线程等待线程池清空太久，又可能超过 max.poll.interval.ms 并触发重平衡。
异步任务尚未结束时，Partition 可能已在重平衡中转移给其他 Consumer。旧 Consumer 中的任务完成后再提交，会遭遇所有权失效、提交失败，甚至与新拥有者的进度发生冲突。
Pod 或进程关闭时，不仅要停止拉取，还要排空、取消或持久化线程池中的在途任务，并在有限停机时间内提交安全位移。
pause() / resume() 实现背压如果业务确实需要异步处理，应设计有界线程池和高低水位：
consumer.pause()；poll()，以驱动消费者组协议并维持存活；consumer.resume()；KafkaConsumer 方法仍只由它的所属线程调用；工作线程只上报处理结果，不能直接并发操作 Consumer；onPartitionsRevoked() 中停止接收对应 Partition 的新任务，并排空、取消或失效化旧任务，然后提交安全水位。pause() 只暂停记录交付，不等于停止调用 poll()。
增加 Partition 并非没有代价：
因此，合理顺序通常是：先优化单条处理和批处理，再评估 Consumer 横向扩容；只有在成本或业务模型要求下，才引入具备完整背压、位移水位和重平衡处理的异步消费架构。
| 配置 | 作用范围 | 核心作用 |
|---|---|---|
segment.bytes / segment.ms | Broker / Topic | 控制日志段滚动 |
retention.ms / retention.bytes | Broker / Topic | 控制删除策略下的数据保留 |
cleanup.policy | Broker / Topic | 选择删除、压缩或二者并用 |
metadata.max.age.ms | Producer | 控制元数据周期刷新 |
batch.size | Producer / Partition | 控制批次大小触发条件 |
linger.ms | Producer | 控制批次主动等待时间 |
buffer.memory | Producer | 控制待发送消息总缓冲预算 |
max.block.ms | Producer 调用线程 | 限制 send() 前置资源等待 |
request.timeout.ms | Producer 单次请求 | 限制一次请求等待响应的时间 |
retry.backoff.ms | Producer Sender | 控制可重试失败后的退避 |
delivery.timeout.ms | Producer 消息生命周期 | 限制总体投递时间 |
acks | Producer | 决定写入确认条件 |
min.insync.replicas | Broker / Topic | 约束 acks=all 时最少同步副本数 |
fetch.min.bytes | Consumer / Broker 响应 | 控制最小拉取响应数据量 |
fetch.max.wait.ms | Consumer / Broker 响应 | 控制 Broker 最长等待时间 |
max.poll.records | Consumer | 限制一次 poll() 返回条数 |
max.poll.interval.ms | Consumer Group | 检测业务处理是否持续推进 |
session.timeout.ms | Consumer Group | 检测 Consumer 是否失联 |
enable.auto.commit | Consumer | 选择自动或手动提交位移 |
auto.offset.reset | Consumer / Partition | 无有效位移时选择起点 |
Comments (0)
Loading comments...