0

我正在使用以下程序在 kafka 中生成记录：

import java.io.IOException;
import java.security.SecureRandom;

public class SensorStatusProducer {
  private final static String TOPIC = "SENSOR_STATUS_DETAILS";
  private final static String PRODUCER_URI = "http://xxx.xxx.xxx.xxx:8082/topics/" + TOPIC;
  private final static SecureRandom randomNumber = new SecureRandom();
  private final static SensorDetails sensorDetails = new SensorDetails();

  public static void main(String[] args) {
    int[] sensorid = sensorDetails.getSensorid(); //this will return [1001,1002,1003,1004,1005]
    try {
      HttpRestProxyUtil rest = new HttpRestProxyUtil(); //this is defined in another class
      for (int sid : sensorid) {
        rest.produceRecords(PRODUCER_URI, String.format("{\"records\":[{\"key\": %d," +
          "\"value\":{" +
          "\"sensorid\":%d," +
          "\"status\":%s," +
          "\"lastconnectedtime\":%s}}]}", sid, sid, "\"CONNECTED\"", String.format("\"%s\"", sensorDetails.currentTimestamp()))); //currentTimestamp() function in defined in another class
      }
    } catch (InterruptedException | IOException me) {
      me.printStackTrace();
    }
  }
}

键的格式说明符为%d，但生成的记录具有 STRING 类型的键。

这可以通过以下方式明显看出：

尝试制作表格时：

CREATE TABLE STATUS_IB_TABLE (ROWKEY INT KEY,
  sensorid INTEGER,
  status VARCHAR,
  lastconnectedtime STRING)
WITH (TIMESTAMP='lastconnectedtime', TIMESTAMP_FORMAT='yyyy-MM-dd HH:mm:ss', KAFKA_TOPIC='SENSOR_STATUS_DETAILS', VALUE_FORMAT='JSON', KEY='sensorid');

在此处输入图像描述

正如@Andrew Coates指出的那样，KEY 被序列化为 STRING

我不知道这怎么可能。

有人可以为我澄清一下，我做错了什么？

PS：

=> 这是我之前的问题ksqlDB 没有正确使用 rowkey的后续问题

=> 融合平台版本：5.5

=> 这是程序的主类。

4

1 回答 1

2

REST 代理支持各种内容类型，但不包括写入序列化 32 位整数的原始类型。

因此，您的代码使用字符串键为主题生成数据。有关如何生成 INT 的示例，请参见此处使用的示例kafkacat。

由于您使用的是 Java，因此您可以使用本机 Java Producer API 来准确控制如何将数据生成到 Kafka（这也比 REST API 更高效、更灵活）。

于 2020-06-02T16:47:05.207 回答