Confluent Cloud 中的消息值未正确解码

Question

我对 kafka 和融合非常陌生。我用我自己的虚拟模型编写了一个与 https://www.confluent.fr/blog/schema-registry-avro-in-spring-boot-application-tutorial/ 上的教程几乎相同的 Producer。 application.yaml 也是一样的。当我将消息发送到 ccloud - 收到的消息是乱码

知道如何解决这个问题吗？当我在发送到 kafka 之前对 avro POJO 执行 System.out.println 时，该对象在所有正确的值下看起来都不错。

{
  "locationId": 1, 
  "time": 1575950400, 
  "temperature": 9.45, 
  "summary": "Overcast", 
  "icon": "cloudy", 
  "precipitationProbability": 0.24,
  ...

而当我从 ccloud 下载消息时，value 看起来像这样

[ 
 {
   "topic":"Weather",
   "partition":0,
   "offset":14,
   "timestamp":1576008230509,
   "timestampType":"CREATE_TIME",
   "headers":[],
   "key":"dummyKey",
   "value":"\u0000\u0000\u0001��\u0002\u0002����\u000b\u0002fffff�\"@\
   ...
}

Answer 1

您实际上做的一切都是正确的 :) 您遇到的只是 Confluent Cloud GUI 在呈现 Avro 消息方面的当前限制。

如果您将消息作为 Avro 使用，您会发现一切正常。下面是一个使用 kafkacat 消费来自 Confluent Cloud 的消息的示例：

$ source .env
$ docker run --rm edenhill/kafkacat:1.5.0 \
          -X security.protocol=SASL_SSL -X sasl.mechanisms=PLAIN \
          -X ssl.ca.location=./etc/ssl/cert.pem -X api.version.request=true \
          -b ${CCLOUD_BROKER_HOST}:9092 \
          -X sasl.username="${CCLOUD_API_KEY}" \
          -X sasl.password="${CCLOUD_API_SECRET}" \
          -r https://"${CCLOUD_SCHEMA_REGISTRY_API_KEY}":"${CCLOUD_SCHEMA_REGISTRY_API_SECRET}"@${CCLOUD_SCHEMA_REGISTRY_HOST} \
          -s avro \
          -t mssql-04-mssql.dbo.ORDERS \
          -f '"'"'Topic %t[%p], offset: %o (Time: %T)\nHeaders: %h\nKey: %k\nPayload (%S bytes): %s\n'"'"' \
          -C -o beginning -c1


Topic mssql-04-mssql.dbo.ORDERS[2], offset: 110 (Time: 1576056196725)
Headers:
Key:
Payload (53 bytes): {"order_id": {"int": 1345}, "customer_id": {"int": 11}, "order_ts": {"int": 18244}, "order_total_usd": {"double": 2.4399999999999999}, "item": {"string": "Bread - Corn Muffaleta Onion"}}

这与此处显示的主题相同，带有二进制 Avro value 字段：