消费渠道 kafka 与 redis 均出现相同问题，直接以 plain => { charset => "BINARY" } 依然出现此问题。

avro 数据原文（ 16 进制）

00000000: 00 80 40 00 cc e1 85 98 0e 00 3a 32 30 32 34 2d  ..@.......:2024-
00000010: 30 34 2d 31 38 54 30 36 3a 33 32 3a 30 34 2e 30  04-18T06:32:04.0
00000020: 30 30 2b 30 38 3a 30 30                          00+08:00

logstash 获取到的内容

00000000: 00 ef bf bd 40 00 ef bf bd ef bf bd ef bf bd ef  ....@...........
00000010: bf bd 0e 00 3a 32 30 32 34 2d 30 34 2d 31 38 54  ....:2024-04-18T
00000020: 30 36 3a 33 32 3a 30 34 2e 30 30 30 2b 30 38 3a  06:32:04.000+08:
00000030: 30 30 0a                                         00.

logstash 是通过 docker 起的 8.13.0 ，以下是容器相关配置

docker-compose.yml

  logstash:
    image: logstash:8.13.0
    volumes:
      - /root/kafka-2-es/logstash.conf:/usr/share/logstash/pipeline/logstash.conf
    environment:
      - "XPACK_MONITORING_ENABLED=false"
      - "KAFKA_GROUP_ID"

logstash.conf

input {
  kafka {
    bootstrap_servers => "kafka:9092"
    topics => ["urllog-test-01"]
    codec => plain {
      charset => "BINARY"
    }
    group_id => "${KAFKA_GROUP_ID}"
    auto_offset_reset => "earliest"
  }
  redis {
    host => "192.168.4.101"
    port => 6379
    password => "Fle7YuG22qIh7ZNPkceopo3oZb1UFZrX"
    data_type => "list"
    key => "urllog-test-01"
    codec => plain {
      charset => "BINARY"
    }
  }
}

output {
  stdout {
    codec => rubydebug
  }
  elasticsearch {
    hosts => ["https://192.168.1.4:9200"]
    index => "urllog-test-01"
    user => "elastic"
    password => "123456"
    ssl => true
    ssl_certificate_verification => false
    manage_template => true
  }
}

已经确认的是:

1. 使用 go 从 kafka 中消费获取到的二进制内容是正确的。
2. 使用 logstash 从 kafka 与 redis 消费相同的内容，均出现特殊字符被替换为 ef bf bd 的情况
3. 使用 kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic urllog-test-01 --from-beginning >> /tmp/data-1.txt 直接从 kafka 中得到的内容是正确的

有没有大佬知道是这么回事，给指条路呗？万分感谢。

第 1 条附言  ·  2024 年 5 月 7 日

input {
  kafka {
    bootstrap_servers => "kafka:9092"
    topics => ["test-01"]
    codec => avro {
      schema_uri => "/app/test.avsc"
      tag_on_failure => true
      encoding => binary
    }
    value_deserializer_class => "org.apache.kafka.common.serialization.ByteArrayDeserializer"
    group_id => "${KAFKA_GROUP_ID}"
    auto_offset_reset => "earliest"
  }
}
output {
  stdout {
    codec => rubydebug
  }
}