Fixed handling of wrong key when text embedded format is used by the …

…producer (#880) Signed-off-by: Paolo Patierno <ppatierno@live.com>
strimzi · Mar 6, 2024 · a3b353d · a3b353d
1 parent 54a9fb1
commit a3b353d
Show file tree

Hide file tree

Showing 2 changed files with 38 additions and 1 deletion.
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/converter/HttpTextMessageConverter.java b/src/main/java/io/strimzi/kafka/bridge/http/converter/HttpTextMessageConverter.java
@@ -38,7 +38,11 @@ public ProducerRecord<byte[], byte[]> toKafkaRecord(String kafkaTopic, Integer p
 
         if (!json.isEmpty()) {
             if (json.has("key")) {
-                key = json.get("key").asText().getBytes();
+                JsonNode keyNode = json.get("key");
+                if (!keyNode.isTextual()) {
+                    throw new IllegalStateException("Because the embedded format is 'text', the key must be a string");
+                }
+                key = keyNode.asText().getBytes();
             }
             if (json.has("value")) {
                 JsonNode valueNode = json.get("value");

diff --git a/src/test/java/io/strimzi/kafka/bridge/http/ProducerIT.java b/src/test/java/io/strimzi/kafka/bridge/http/ProducerIT.java
@@ -540,6 +540,39 @@ void sendTextMessageWithWrongValue(VertxTestContext context) throws InterruptedE
                 });
     }
 
+    @Test
+    void sendTextMessageWithWrongKey(VertxTestContext context) throws InterruptedException, ExecutionException {
+        KafkaFuture<Void> future = adminClientFacade.createTopic(topic);
+
+        JsonObject key = new JsonObject().put("my-key", "This is a json key");
+
+        JsonArray records = new JsonArray();
+        JsonObject json = new JsonObject();
+        json.put("key", key);
+        json.put("value", "Text value");
+        records.add(json);
+
+        JsonObject root = new JsonObject();
+        root.put("records", records);
+
+        future.get();
+
+        // produce and check the status code
+        producerService()
+                .sendRecordsRequest(topic, root, BridgeContentType.KAFKA_JSON_TEXT)
+                .sendJsonObject(root, ar -> {
+                    context.verify(() -> {
+                        assertThat(ar.succeeded(), is(true));
+                        HttpResponse<JsonObject> response = ar.result();
+                        HttpBridgeError error = HttpBridgeError.fromJson(response.body());
+                        assertThat(response.statusCode(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
+                        assertThat(error.getCode(), is(HttpResponseStatus.UNPROCESSABLE_ENTITY.code()));
+                        assertThat(error.getMessage(), is("Because the embedded format is 'text', the key must be a string"));
+                    });
+                    context.completeNow();
+                });
+    }
+
     @Test
     void sendMessageWithNullValueTest(VertxTestContext context) throws InterruptedException, ExecutionException {
         String topic = "sendMessageWithNullValueTest";