虽然REST程序有INT，但是Kafka将消息键作为字符串生成？

Question

我用以下程序制作卡夫卡的唱片：

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，但生成的记录具有字符串类型。

从以下几个方面可以看出这一点：

当试图制作桌子时：

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所指出的，键被序列化为字符串。

我不知道这怎么可能。

有人能帮我澄清一下吗，我做错什么了？

PS：

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

=>汇合平台版本: 5.5

=> --这是程序的主要类。

Answer 1

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

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

由于您使用的是Java，所以您可以使用原生Java生产者API来精确地控制数据是如何生成给Kafka的(与REST相比，Kafka的性能和灵活性也更高)。