在kafka res api响应中获取垃圾值

Question

如果来自融合云上的 kafka 实例，我正在使用 rest api 获取主题

我正在使用下面的 curl 命令

curl "http://(myhost):9092/topics/(topicname)" --output op.txt

但我在op.txt 中得到了垃圾值

"U^C^C^@^B^BP"

有什么解决办法吗？

Answer 1

您不能使用 REST 从 Confluent Cloud 消费（目前）。使用 Confluent Cloud 中的数据，您可以使用 ccloud 或 kafkacat 等工具从命令行访问您的数据。

Confluent Cloud CLI (ccloud)

$ ccloud kafka topic consume --from-beginning rmoff_test_topic_01
Starting Kafka Consumer. ^C or ^D to exit
Hello world!
This is a message on a topic in Confluent Cloud

卡夫卡猫

您可以在本地运行 kafkacat，也可以从 Docker 中使用它。

docker run --rm --interactive edenhill/kafkacat:1.6.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 \
            -X sasl.username="$CCLOUD_API_KEY" \
            -X sasl.password="$CCLOUD_API_SECRET" \
            -t rmoff_test_topic_01 -C -u -e
Hello world!
This is a message on a topic in Confluent Cloud

您还可以以 JSON 格式输出消息，这很有用：

docker run --rm --interactive edenhill/kafkacat:1.6.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 \
            -X sasl.username="$CCLOUD_API_KEY" \
            -X sasl.password="$CCLOUD_API_SECRET" \
            -t rmoff_test_topic_01 -C -u -J -e

{
  "topic": "rmoff_test_topic_01",
  "partition": 0,
  "offset": 0,
  "tstype": "create",
  "ts": 1604571163960,
  "broker": 7,
  "key": null,
  "payload": "Hello world!"
}
{
  "topic": "rmoff_test_topic_01",
  "partition": 3,
  "offset": 0,
  "tstype": "create",
  "ts": 1604571168723,
  "broker": 1,
  "key": null,
  "payload": "This is a message on a topic in Confluent Cloud"
}