feat: added a new text embedded format

src/main/java/io/strimzi/kafka/bridge/BridgeContentType.java

@@ -16,9 +16,16 @@ public class BridgeContentType {
     /** JSON encoding with BINARY embedded format */
     public static final String KAFKA_JSON_BINARY = "application/vnd.kafka.binary.v2+json";
 
+
+    /** JSON encoding with Plain text embedded format */
+    public static final String KAFKA_JSON_TEXT = "application/vnd.kafka.string.v2+json";
+
     /** Specific Kafka JSON encoding */
     public static final String KAFKA_JSON = "application/vnd.kafka.v2+json";
 
     /** JSON encoding */
     public static final String JSON = "application/json";
+
+    /** Plain text encoding */
+    public static final String TEXT = "text/plain";
 }

src/main/java/io/strimzi/kafka/bridge/EmbeddedFormat.java

@@ -10,11 +10,20 @@
  */
 public enum EmbeddedFormat {
 
-    /** Define "binary" data as embedded format */
+    /**
+     * Define "binary" data as embedded format
+     */
     BINARY,
 
-    /** Define "json" data as embedded format */
-    JSON;
+    /**
+     * Define "json" data as embedded format
+     */
+    JSON,
+
+    /**
+     * Define "text" data as embedded format
+     */
+    TEXT;
 
     /**
      * Convert the String value in the corresponding enum
@@ -28,6 +37,8 @@ public static EmbeddedFormat from(String value) {
                 return JSON;
             case "binary":
                 return BINARY;
+            case "text":
+                return TEXT;
         }
         throw new IllegalEmbeddedFormatException("Invalid format type.");
     }

src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java

@@ -22,6 +22,7 @@
 import io.strimzi.kafka.bridge.http.converter.HttpJsonMessageConverter;
 import io.strimzi.kafka.bridge.http.converter.JsonDecodeException;
 import io.strimzi.kafka.bridge.http.converter.JsonUtils;
+import io.strimzi.kafka.bridge.http.converter.HttpTextMessageConverter;
 import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
 import io.strimzi.kafka.bridge.tracing.SpanHandle;
 import io.strimzi.kafka.bridge.tracing.TracingHandle;
@@ -74,10 +75,10 @@ public class HttpSinkBridgeEndpoint<K, V> extends HttpBridgeEndpoint {
     /**
      * Constructor
      *
-     * @param bridgeConfig the bridge configuration
-     * @param context the HTTP bridge context
-     * @param format the embedded format for consumed messages
-     * @param keyDeserializer key deserializer for consumed messages
+     * @param bridgeConfig      the bridge configuration
+     * @param context           the HTTP bridge context
+     * @param format            the embedded format for consumed messages
+     * @param keyDeserializer   key deserializer for consumed messages
      * @param valueDeserializer value deserializer for consumed messages
      */
     public HttpSinkBridgeEndpoint(BridgeConfig bridgeConfig, HttpBridgeContext<K, V> context, EmbeddedFormat format,
@@ -111,8 +112,8 @@ public void close() {
      * Create a Kafka consumer
      *
      * @param routingContext the routing context
-     * @param bodyAsJson HTTP request body bringing consumer settings
-     * @param handler handler for the request
+     * @param bodyAsJson     HTTP request body bringing consumer settings
+     * @param handler        handler for the request
      */
     private void doCreateConsumer(RoutingContext routingContext, JsonNode bodyAsJson, Handler<HttpBridgeEndpoint> handler) {
         // get the consumer group-id
@@ -528,7 +529,7 @@ public void doUnsubscribe(RoutingContext routingContext) {
     /**
      * Add a configuration parameter with key and value to the provided Properties bag
      *
-     * @param key key of the configuration parameter
+     * @param key   key of the configuration parameter
      * @param value value of the configuration parameter
      * @param props Properties bag where to put the configuration parameter
      */
@@ -608,6 +609,8 @@ private MessageConverter<K, V, byte[], byte[]> buildMessageConverter() {
                 return (MessageConverter<K, V, byte[], byte[]>) new HttpJsonMessageConverter();
             case BINARY:
                 return (MessageConverter<K, V, byte[], byte[]>) new HttpBinaryMessageConverter();
+            case TEXT:
+                return (MessageConverter<K, V, byte[], byte[]>) new HttpTextMessageConverter();
         }
         return null;
     }
@@ -618,6 +621,8 @@ private boolean checkAcceptedBody(String accept) {
                 return format == EmbeddedFormat.JSON;
             case BridgeContentType.KAFKA_JSON_BINARY:
                 return format == EmbeddedFormat.BINARY;
+            case BridgeContentType.KAFKA_JSON_TEXT:
+                return format == EmbeddedFormat.TEXT;
         }
         return false;
     }

src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java

@@ -16,6 +16,7 @@
 import io.strimzi.kafka.bridge.converter.MessageConverter;
 import io.strimzi.kafka.bridge.http.converter.HttpBinaryMessageConverter;
 import io.strimzi.kafka.bridge.http.converter.HttpJsonMessageConverter;
+import io.strimzi.kafka.bridge.http.converter.HttpTextMessageConverter;
 import io.strimzi.kafka.bridge.http.converter.JsonUtils;
 import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
 import io.strimzi.kafka.bridge.http.model.HttpBridgeResult;
@@ -114,7 +115,7 @@ public void handle(RoutingContext routingContext, Handler<HttpBridgeEndpoint> ha
             }
             records = messageConverter.toKafkaRecords(topic, partition, routingContext.body().buffer().getByteBuf().array());
 
-            for (ProducerRecord<K, V> record :records)   {
+            for (ProducerRecord<K, V> record : records) {
                 span.inject(record);
             }
         } catch (Exception e) {
@@ -216,6 +217,8 @@ private MessageConverter<K, V, byte[], byte[]> buildMessageConverter() {
                 return (MessageConverter<K, V, byte[], byte[]>) new HttpJsonMessageConverter();
             case BINARY:
                 return (MessageConverter<K, V, byte[], byte[]>) new HttpBinaryMessageConverter();
+            case TEXT:
+                return (MessageConverter<K, V, byte[], byte[]>) new HttpTextMessageConverter();
         }
         return null;
     }

src/main/java/io/strimzi/kafka/bridge/http/converter/HttpTextMessageConverter.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.kafka.bridge.http.converter;
+
+import io.strimzi.kafka.bridge.converter.MessageConverter;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+
+import java.util.List;
+
+/**
+ * Implementation of a message converter to deal with the "text" embedded data format
+ */
+public class HttpTextMessageConverter implements MessageConverter<byte[], byte[], byte[], byte[]> {
+
+    Integer partitionFromBody = null;
+    byte[] key = null;
+    byte[] value = null;
+    Headers headers = new RecordHeaders();
+
+    @Override
+    public ProducerRecord<byte[], byte[]> toKafkaRecord(String kafkaTopic, Integer partition, byte[] message) {
+        return null;
+    }
+
+    @Override
+    public List<ProducerRecord<byte[], byte[]>> toKafkaRecords(String kafkaTopic, Integer partition, byte[] messages) {
+        return null;
+    }
+
+    @Override
+    public byte[] toMessage(String address, ConsumerRecord<byte[], byte[]> record) {
+        return new byte[0];
+    }
+
+    @Override
+    public byte[] toMessages(ConsumerRecords<byte[], byte[]> records) {
+        return new byte[0];
+    }
+}