kubernetes 滚动更新时，kafka 消费者因进程被强制终止而未完成消息处理与 offset 提交，导致消息丢失；需通过 sigterm 信号捕获 + 优雅停机 + 合理的 terminationgraceperiodseconds 配置来彻底解决。
kubernetes 滚动更新时，kafka 消费者因进程被强制终止而未完成消息处理与 offset 提交，导致消息丢失；需通过 sigterm 信号捕获 + 优雅停机 + 合理的 terminationgraceperiodseconds 配置来彻底解决。
在基于 BasicKafkaConsumerV2 构建的消费者服务中，尽管已禁用自动提交（enable_auto_commit=False）并采用手动 commit()，仍于 Pod 重启期间出现消息丢失，根本原因并非 Kafka 协议缺陷，而是应用层缺乏对容器生命周期事件的响应能力。
Kubernetes 在滚动部署时会向容器主进程发送 SIGTERM 信号，并在默认 30 秒（可通过 terminationGracePeriodSeconds 调整）后强制执行 SIGKILL。若消费者未监听 SIGTERM，则可能在以下任一环节中断：
- 正在处理某条消息（如 DB 写入中）；
- 已调用
message_handler_wrapped但尚未执行self.consumer.commit()； - 已提交 offset，但日志或业务逻辑尚未完成（造成“看似丢失”假象）。
因此，必须实现可中断、可恢复、可确认的优雅停机流程。
✅ 正确做法：注册 SIGTERM 处理器 + 原子化消费循环
首先，在消费者初始化后立即注册信号处理器：
import signal
import sys
def graceful_shutdown(signum, frame):
logger.info(f"[{self.consumer_name}] Received SIGTERM, initiating graceful shutdown...")
# 1. 停止拉取消息（中断 for msg in self.consumer 循环）
if hasattr(self, 'consumer') and self.consumer:
self.consumer.close() # 触发 KafkaConsumer.__iter__ 抛出 StopIteration
# 2. 确保最后一批消息完成处理并提交（如有待处理 msg）
# （注意：此处需配合循环外状态管理，见下文）
logger.info(f"[{self.consumer_name}] Shutdown completed.")
sys.exit(0)
# 在 __init__ 或 start_consumer 开头注册
signal.signal(signal.SIGTERM, lambda s, f: graceful_shutdown(s, f))但更关键的是重构 start_consumer()，避免阻塞式无限循环导致无法响应信号：
def start_consumer(self):
logger.info(f"Consumer [{self.consumer_name}] is starting consuming")
try:
for msg in self.consumer:
with LogGuidSetter():
self.message_handler_wrapped(msg.topic, msg.value, msg.headers, msg)
self.consumer.commit()
logger.info(
f"[{self.consumer_name}] Consumed message from partition: {msg.partition} "
f"offset: {msg.offset} with key: {msg.key}"
)
except KeyboardInterrupt:
logger.info(f"[{self.consumer_name}] Keyboard interrupt received.")
except Exception as e:
logger.exception(f"[{self.consumer_name}] Unexpected error in consumer loop: {e}")
finally:
# 确保无论何种退出，都尝试关闭 consumer
if hasattr(self, 'consumer') and self.consumer:
try:
self.consumer.close()
logger.info(f"[{self.consumer_name}] Kafka consumer closed gracefully.")
except Exception as close_err:
logger.error(f"[{self.consumer_name}] Failed to close consumer: {close_err}")⚠️ 注意：
KafkaConsumer.close()会中断for msg in self.consumer迭代，使循环自然退出，这是响应SIGTERM的核心机制。
✅ Kubernetes 层配置强化
在 Deployment YAML 中显式设置合理的终止宽限期（推荐 ≥ 60s），确保应用有足够时间完成当前消息：
apiVersion: apps/v1
kind: Deployment
metadata:
name: order-consumer
spec:
template:
spec:
terminationGracePeriodSeconds: 60 # 关键！给 Python 留足 commit 时间
containers:
- name: order-consumer
image: KUSTOMIZE_PRIMARY
command:
- "/wait-for.sh"
- "localhost:6432"
- "-s"
- "-t"
- "30"
- "--"
- "ddtrace-run"
- "python"
- "manage.py"
- "run_order-consumer"✅ 补充建议：增强可靠性
-
启用
session.timeout.ms与heartbeat.interval.ms：避免因 GC 或短暂卡顿触发误判的 rebalance（例如设为session.timeout.ms=45000,heartbeat.interval.ms=15000）； - 日志中记录 commit 前后 offset：便于审计是否真丢失，而非仅“未打印日志”；
-
考虑使用
commit_async()+ callback 做失败重试（适用于高吞吐场景，但需注意线程安全）； -
DLQ（死信队列）兜底：在
except Exception分支中将异常消息转发至 DLQ Topic，而非静默丢弃。
通过信号捕获、消费循环可中断设计、K8s 宽限期协同，即可彻底消除滚动部署中的消息丢失问题——这不仅是 Kafka 最佳实践，更是云原生应用健壮性的基本要求。
