如何在Logstash中获取Kafka Producer Header里的eventType值？

问题核心 #

Kafka Producer发送的Header值在Logstash中以字节数组形式存在，直接引用会得到字节的数值数组而非字符串，这是提取失败的主要原因。

前置确认 #

先通过Logstash标准输出生成器验证Header是否被正确捕获：

input { kafka { ... } }
output { stdout { codec => rubydebug } }

查看输出中[@metadata][kafka][headers]的结构，确认eventType键存在，值为类似[85,115,101,114,67,114,101,97,116,101,100]的字节数组。

可行解决方案 #

使用ruby过滤器将字节数组转换为UTF-8字符串，完整配置示例如下：

完整Logstash配置 #

input {
  kafka {
    bootstrap_servers => "your-kafka-broker:9092"
    topics => ["your-multi-event-topic"]
    auto_offset_reset => "latest"
    decorate_events => true # 必须开启，才能在[@metadata][kafka]中获取Headers
  }
}

filter {
  # 转换eventType Header为字符串
  ruby {
    code => '
      event_type_bytes = event.get("[@metadata][kafka][headers][eventType]")
      # 仅当字节数组存在时执行转换
      if event_type_bytes.is_a?(Array) && !event_type_bytes.empty?
        event.set("eventType", event_type_bytes.pack("C*").force_encoding("UTF-8"))
      end
    '
  }

  # 可选：根据eventType路由到不同ES索引
  if [eventType] == "UserCreated" {
    mutate { add_field => { "[@metadata][index]" => "user-events-%{+YYYY.MM.dd}" } }
  } elsif [eventType] == "OrderPlaced" {
    mutate { add_field => { "[@metadata][index]" => "order-events-%{+YYYY.MM.dd}" } }
  }
}

output {
  elasticsearch {
    hosts => ["your-es-node:9200"]
    index => "%{[@metadata][index]}" # 使用动态索引
    # 其他ES配置（如认证）
  }
}

关键细节说明 #

1. decorate_events必须设为true：只有开启这个配置，Kafka的元数据（包括Headers）才会被写入[@metadata][kafka]路径下。
2. 字节数组转字符串的逻辑：
   • pack("C*")将字节数组（每个元素是0-255的整数）转换为原始字节序列
   • force_encoding("UTF-8")确保序列被解析为UTF-8编码的字符串
3. 大小写敏感：确保Logstash中引用的Header键名（eventType）与C# Producer中设置的完全一致（Kafka Header键名区分大小写）。

常见错误排查 #

• 如果提取后eventType字段为空：检查Header键名拼写、大小写是否匹配，或Producer是否确实发送了该Header。
• 如果出现编码乱码：确认Producer发送Header时使用的是UTF-8编码（如C#中的Encoding.UTF8.GetBytes()）。

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