ksqldb - ksql rest api 创建并在 json 中获取输出

Question

我无法从 katable 获取数据到 pojo 类输出

我试图调用融合的 kafka 的休息端点，但我无法将输出放在 pojo 类中。

    @PostMapping("/QueryForEquipment")
    @Consumes("application/json")
    public String getEquipments() {
            String walletBalanceUrl = "http://172.21.79.18:8088/query";

            HttpHeaders httpHeaders = new HttpHeaders();
            httpHeaders.set("Content-Type", "application/json");
            httpHeaders.add("Accept", MediaType.APPLICATION_JSON.toString());

            JSONObject json = new JSONObject();
            json.put("ksql", "Select * from  EQP_STREAM limit 10;");

            JSONObject jsonSub = new JSONObject();
            jsonSub.put("ksql.streams.auto.offset.reset", "earliest");
            json.put("streamsProperties", jsonSub);

         /*   String body=new String("{
                    "ksql": "Select ROWTIME,ROWKEY,TRN_ID_KEY from TABLE_EQP_LOCATION limit 10;",
                "streamsProperties": {"ksql.streams.auto.offset.reset": "earliest"} }"); 

         */

            System.out.println(walletBalanceUrl);

            HttpEntity httpEntity = new HttpEntity(body, httpHeaders);

            RestTemplate restTemplate = new RestTemplate();
            System.out.println("json text====" + body);

            ResponseEntity<String> uri = restTemplate.postForEntity(walletBalanceUrl, httpEntity, String.class);

            return uri.getBody().toString();

预期输出必须是json格式目前它作为

{"row":{"columns":[1556458915675,"CDAU603054","Q1214320190418","TRAIN","Arrived","BELLEVILLE","Q","1214","4",1555545600000," ",1555618200000,"SRS_TRAIN","-77.374856","44.179657",null,null,"CDAU603054","CDAU","603054","INTERMODAL","L","KC7","7","DTTX720029","DTTX","720029","C","T","TOP",1555621693617]},"errorMessage":null,"finalMessage":null}
{"row":{"columns":[1556458915680,"CDAU603066","Q1214320190418","TRAIN","Arrived","BELLEVILLE","Q","1214","4",1555545600000," ",1555618200000,"SRS_TRAIN","-77.374856","44.179657",null,null,"CDAU603066","CDAU","603066","INTERMODAL","L","KC7","8","DTTX720030","DTTX","720030","D","T","TOP",1555621693617]},"errorMessage":null,"finalMessage":null}
{"row":{"columns":[1556458915680,"CDAU603070","Q1214320190418","TRAIN","Arrived","BELLEVILLE","Q","1214","4",1555545600000," ",1555618200000,"SRS_TRAIN","-77.374856","44.179657",null,null,"CDAU603070","CDAU","603070","INTERMODAL","L","KC7","8","DTTX720030","DTTX","720030","A","T","TOP",1555621693617]},"errorMessage":null,"finalMessage":null}

Answer 1

在这里，Content-Type 将是 application/vnd.ksql.v1+json 并设置charset 到 utf-8。

    HttpHeaders headers = new HttpHeaders();
    headers.set("Content-Type", "application/vnd.ksql.v1+json");
    headers.set("charset", "utf-8");
    headers.setAccept(Collections.singletonList(MediaType.APPLICATION_JSON));

ksql 查询将是SELECT * from EQP_STREAM EMIT CHANGES limit 10;

     json.put("ksql", "SELECT * from  EQP_STREAM EMIT CHANGES limit 10;")

引用链接: https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-rest-api/ksql-endpoint/