Kiba ETL调用Faraday时无报错静默退出问题排查求助

核心问题 #

基于CDC架构的项目中，Debezium捕获数据库变更推送到RabbitMQ，Kiba ETL消费消息时，RetrieveJobDataTransform类调用Faraday发起API请求后，流水线无报错静默终止，但该类在IRB中单独运行完全正常。

排查步骤 #

1. 强制捕获并输出所有异常 #

Kiba默认可能吞掉部分底层异常，需在关键方法中添加全局异常捕获，确保错误信息被打印：

def process(row)
  puts "\t[transform] Retrieve Job Information by UUID - job_uuid: #{row}"
  raise ArgumentError, 'row cannot be nil' if row.nil?

  begin
    job_uuid = row.dig(:after_data, 'job_uuid')
    api_result = make_api_call(job_uuid)
    transformed_data = transform_data(row, api_result)
    transformed_data
  rescue => e
    puts "[ERROR] Transform failed: #{e.message}"
    puts e.backtrace.join("\n")
    # 根据需求选择是否终止流程，或返回原数据继续
    raise e
  end
end

private

def make_api_call(job_uuid)    
  begin
    conn = create_connection('https://url.com')    
    puts "\t[debug] About to call API with job_uuid: #{job_uuid}"
    response = conn.get("api/jobs/#{job_uuid}")
    puts "\t[debug] API response status: #{response.status}"

    if response.success?
      JSON.parse(response.body)
    else
      nil
    end
  rescue => e
    puts "[ERROR] API call failed: #{e.message}"
    puts e.backtrace.join("\n")
    raise e
  end
end

2. 排查Faraday响应中间件冲突 #

你同时配置了f.response :json和手动调用JSON.parse(response.body)，可能存在解析逻辑冲突：

• 尝试去掉手动解析，直接使用Faraday中间件解析后的response.body（已转为Hash）
• 或暂时注释f.response :json，保留手动解析，验证是否还会静默退出

3. 对比IRB与Kiba的运行环境差异 #

• 环境变量：检查API密钥、代理配置等环境变量在Kiba运行时是否与IRB一致
• 依赖版本：执行bundle list faraday，与IRB中Faraday.version对比，确保版本无差异
• Ruby版本：确认Kiba和IRB使用的Ruby版本完全相同

4. 启用Kiba调试日志 #

开启Kiba的调试日志，查看内部执行流程：

Kiba.run(job, logger: Logger.new(STDOUT, level: Logger::DEBUG))

5. 测试API调用的网络与超时问题 #

虽然设置了120秒超时，但网络层面的问题（如DNS失败、TCP连接重置）可能直接终止进程：

• 替换API地址为测试接口（如http://httpbin.org/get），验证是否还会退出
• 开启Faraday请求日志，查看完整的请求响应细节：

def create_connection(api_url)
  Faraday.new(api_url) do |f|
    f.headers['x-api-key'] = 'api key'
    f.response :json 
    f.adapter :net_http 
    f.options.timeout = 120
    f.response :logger, Logger.new(STDOUT), bodies: true # 打印请求响应内容
  end
end

6. 检查RabbitMQ消费的线程模型 #

如果Kiba的RabbitMQ Source基于线程实现，可能与Faraday存在线程安全冲突：

• 暂时将Source替换为静态测试数据，排除RabbitMQ线程的影响
• 确认每次API调用都创建新的Faraday连接（当前代码已实现，无需修改）

7. 跟踪进程终止信号 #

在Linux/macOS下用strace跟踪进程，查看是否收到终止信号：

strace -f -tt ruby your_kiba_script.rb

额外优化点 #

transform_data方法中若api_result为nil，会导致后续流程拿到空数据，建议添加兜底逻辑：

def transform_data(row, api_result)
  api_result || row # API调用失败时返回原数据，避免流程中断
end

内容的提问来源于stack exchange，提问作者Edson Lek Hong Ma