fixed no schema matches error (strimzi#895)

* fixed no schema matches error Signed-off-by: Antonio Pedro <[email protected]> * Added missings defintions and improve the test Signed-off-by: Antonio Pedro <[email protected]> * improve CHANGELOG.md Signed-off-by: António Pedro <[email protected]> * Update CHANGELOG.md Co-authored-by: Paolo Patierno <[email protected]> Signed-off-by: António Pedro <[email protected]> --------- Signed-off-by: Antonio Pedro <[email protected]> Signed-off-by: António Pedro <[email protected]> Co-authored-by: Paolo Patierno <[email protected]>
PaulRMellor · Apr 26, 2024 · 86783da · 86783da
1 parent 3bf3818
commit 86783da
Show file tree

Hide file tree

Showing 4 changed files with 119 additions and 0 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -3,6 +3,7 @@
 ## 0.29.0
 
 * Dependency updates (Vert.x 4.5.7, Netty 4.1.108.Final)
+* Added support for records key and value to be JSON array when using the `json` embedded format.
 
 ## 0.28.0
 

diff --git a/src/main/resources/openapi.json b/src/main/resources/openapi.json
@@ -1494,6 +1494,9 @@
                 "properties": {
                     "key": {
                         "oneOf": [
+                            {
+                                "type": "array"
+                            },
                             {
                                 "type": "object"
                             },
@@ -1809,6 +1812,9 @@
                     },
                     "value": {
                         "oneOf": [
+                            {
+                                "type": "array"
+                            },
                             {
                                 "type": "object",
                                 "nullable": true
@@ -1821,6 +1827,9 @@
                     },
                     "key": {
                         "oneOf": [
+                            {
+                                "type": "array"
+                            },
                             {
                                 "type": "object"
                             },
@@ -1887,6 +1896,9 @@
                 "properties": {
                     "value": {
                         "oneOf": [
+                            {
+                                "type": "array"
+                            },
                             {
                                 "type": "object"
                             },
@@ -1897,6 +1909,9 @@
                     },
                     "key": {
                         "oneOf": [
+                            {
+                                "type": "array"
+                            },
                             {
                                 "type": "object"
                             },

diff --git a/src/main/resources/openapiv2.json b/src/main/resources/openapiv2.json
@@ -1331,6 +1331,7 @@
       "properties": {
         "key": {
           "type": [
+            "array",
             "object",
             "string"
           ]
@@ -1348,6 +1349,7 @@
         },
         "value": {
           "type": [
+            "array",
             "object",
             "string",
             "null"
@@ -1626,13 +1628,15 @@
         },
         "value": {
           "type": [
+            "array",
             "object",
             "string",
             "null"
           ]
         },
         "key": {
           "type": [
+            "array",
             "object",
             "string"
           ]
@@ -1695,12 +1699,14 @@
       "properties": {
         "value": {
           "type": [
+            "array",
             "object",
             "string"
           ]
         },
         "key": {
           "type": [
+            "array",
             "object",
             "string"
           ]

diff --git a/src/test/java/io/strimzi/kafka/bridge/http/ProducerIT.java b/src/test/java/io/strimzi/kafka/bridge/http/ProducerIT.java
@@ -201,6 +201,103 @@ void sendSimpleMessageWithKey(VertxTestContext context) throws InterruptedExcept
         });
     }
 
+    @Test
+    void sendSimpleMessangeWithArrayKey(VertxTestContext context) throws InterruptedException, ExecutionException {
+        KafkaFuture<Void> future = adminClientFacade.createTopic(topic, 2, 1);
+
+        String value = "message-value";
+        JsonArray key = new JsonArray();
+        key.add("some-element").add(new JsonObject().put("some-field", "element-2"));
+
+        JsonArray records = new JsonArray();
+        JsonObject json = new JsonObject();
+        json.put("value", value);
+        json.put("key", key);
+        records.add(json);
+
+        JsonObject root = new JsonObject();
+        root.put("records", records);
+
+        future.get();
+
+        producerService()
+            .sendRecordsRequest(topic, root, BridgeContentType.KAFKA_JSON_JSON)
+            .sendJsonObject(root, verifyOK(context));
+
+        Properties consumerProperties = Consumer.fillDefaultProperties();
+        consumerProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaUri);
+
+        KafkaConsumer<byte[], String> consumer = KafkaConsumer.create(vertx, consumerProperties,
+                new ByteArrayDeserializer(), new KafkaJsonDeserializer<>(String.class));
+        consumer.handler(record -> {
+            context.verify(() -> {
+                assertThat(record.value(), is(value));
+                assertThat(record.topic(), is(topic));
+                assertThat(record.partition(), notNullValue());
+                assertThat(record.offset(), is(0L));
+                assertThat(record.key(), is(key.toBuffer().getBytes()));
+            });
+            LOGGER.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
+                    record.topic(), record.partition(), record.offset(), record.key(), record.value());
+            consumer.close();
+            context.completeNow();
+        });
+
+        consumer.subscribe(topic, done -> {
+            if (!done.succeeded()) {
+                context.failNow(done.cause());
+            }
+        });
+    }
+
+
+    @Test
+    void sendSimpleMessageWithArrayValue(VertxTestContext context) throws InterruptedException, ExecutionException {
+        KafkaFuture<Void> future = adminClientFacade.createTopic(topic, 2, 1);
+
+        JsonArray value = new JsonArray();
+        value.add("some-element").add(new JsonObject().put("some-field", "element-2"));
+
+        JsonArray records = new JsonArray();
+        JsonObject json = new JsonObject();
+        json.put("value", value);
+        records.add(json);
+
+        JsonObject root = new JsonObject();
+        root.put("records", records);
+
+        future.get();
+
+        producerService()
+            .sendRecordsRequest(topic, root, BridgeContentType.KAFKA_JSON_JSON)
+            .sendJsonObject(root, verifyOK(context));
+
+        Properties consumerProperties = Consumer.fillDefaultProperties();
+        consumerProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaUri);
+
+        KafkaConsumer<String, byte[]> consumer = KafkaConsumer.create(vertx, consumerProperties,
+                new StringDeserializer(), new ByteArrayDeserializer());
+        consumer.handler(record -> {
+            context.verify(() -> {
+                assertThat(record.value(), is(value.toBuffer().getBytes()));
+                assertThat(record.topic(), is(topic));
+                assertThat(record.partition(), notNullValue());
+                assertThat(record.offset(), is(0L));
+                assertThat(record.key(), nullValue());
+            });
+            LOGGER.info("Message consumed topic={} partition={} offset={}, key={}, value={}",
+                    record.topic(), record.partition(), record.offset(), record.key(), record.value());
+            consumer.close();
+            context.completeNow();
+        });
+
+        consumer.subscribe(topic, done -> {
+            if (!done.succeeded()) {
+                context.failNow(done.cause());
+            }
+        });
+    }
+
     @Disabled("Will be check in the next PR, this is just external tests for Bridge")
     @DisabledIfEnvironmentVariable(named = "EXTERNAL_BRIDGE", matches = "((?i)TRUE(?-i))")
     @Test