AWS Fargate+MSK环境下Karafka单条消费时进程挂起求助

配置详情 #

Karafka配置及代码 #

class KarafkaApp < Karafka::App
  setup do |config|
    config.kafka = {
      'bootstrap.servers': 'brokers',
      'enable.partition.eof': true,
      'security.protocol': 'sasl_ssl',
      'sasl.mechanism': 'SCRAM-SHA-512',
      'sasl.username': 'username',
      'sasl.password': 'password',
    }
    config.client_id = 'users_and_groups_service'
    config.max_messages = 1
    config.consumer_persistence = !Rails.env.development?
    config.logger = Logger.new(STDOUT)
    config.logger.level = Logger::DEBUG
  end

  routes.draw do
    topic 'my_topic' do
      consumer MyConsumer
    end
  end
end

class SingleMessageBaseConsumer < Karafka::BaseConsumer
  attr_reader :message

  def consume
    messages.each do |message|
      @message = message
      consume_one
    
      mark_as_consumed(message)
    end
  end
end

class MyConsumer < SingleMessageBaseConsumer
  def consume_one
    Karafka.logger.info 'starting'
    sleep 3
    Karafka.logger.info 'finished'
  end
end

容器中librdkafka的安装配置 #

wget https://github.com/edenhill/librdkafka/archive/refs/tags/v2.3.0.tar.gz && \
tar -xzf v2.3.0.tar.gz && \
cd librdkafka-2.3.0 && \
./configure --enable-sasl --enable-ssl && \
make && \
make install

问题现象 #

• 本地容器化Kafka环境运行正常，部署到AWS Fargate后出现异常：
  • 消息业务逻辑执行需1-2秒时，约1秒后日志突然停止，进程冻结，需发送更多消息或重启容器才能恢复；
  • 简化消费者仅保留sleep和日志时，单条消息无日志输出，多条消息处理中途日志中断。
• 排除单纯批量消费等待问题，怀疑与环境或配置相关，此前使用Ruby Racecar也遇到相同问题。

疑问 #

1. 有没有人在AWS Fargate+Amazon MSK环境下遇到过Karafka的类似问题？
2. 该问题是否与环境中librdkafka的编译或配置有关？

排查思路与解决方案 #

1. 检查librdkafka编译与MSK兼容性 #

• 编译librdkafka时明确添加--enable-sasl-scram参数，确保SCRAM机制被正确启用，避免自动检测遗漏；
• 替换手动编译方式，使用操作系统官方预编译包（如Alpine的apk add librdkafka-dev、Debian的apt-get install librdkafka-dev），消除手动编译的环境差异；
• 核对MSK集群版本与librdkafka的兼容列表，使用MSK官方推荐的librdkafka版本（当前推荐2.0+）。

2. 调整Karafka与Kafka消费者配置 #

• 增加Kafka会话超时配置：添加'session.timeout.ms': 30000和'heartbeat.interval.ms': 10000，适配Fargate网络延迟，避免消费者被误判离线；
• 关闭enable.partition.eof：将该配置设为false，避免单消息消费完成后消费者进入EOF等待状态；
• 显式设置'fetch.min.bytes': 1，确保消费者立即获取可用消息，无需等待凑齐批量；
• 临时禁用consumer_persistence测试，排查是否是持久化消费者导致的连接异常。

3. Fargate环境相关排查 #

• 检查任务资源配置：确认Fargate任务的CPU/内存足够，查看CloudWatch日志是否有OOM（内存不足）终止事件；
• 验证网络连通性：确保Fargate安全组允许访问MSK的9096端口，MSK安全组也开放对应权限给Fargate；
• 强制刷新日志输出：在日志语句后添加STDOUT.flush，避免因Fargate日志缓冲导致的“冻结”假象；
• 开启Kafka调试日志：在config.kafka中添加'debug': 'consumer,topic,fetch'，获取消费者交互细节，定位冻结阶段。

4. 代码逻辑优化 #

• 简化消息处理逻辑：因max_messages=1，consume方法无需遍历消息，直接使用messages.first处理，减少冗余代码；
• 确认偏移量提交时机：处理完成后立即调用mark_as_consumed，避免偏移量未提交导致的状态异常。

内容的提问来源于stack exchange，提问作者user2631151