feat: support for schemas in JSON record builder

This commit introduces support for emitting structured records
from the JSON record builder. This will allow the MQ Source
Connector to read JSON string messages from MQ, and produce them
to Kafka using any standard Converter (e.g. to produce them in
Avro or Protobuf formats if desired).

The JsonConverter dependency used in JSON record builder has
support for this from Kafka Connect v4.2, so the simplest
implementation would be to update the dependency in pom.xml
to version 4.2, and just pass through the schemas.enable and
schema.content configuration properties to the converter
and leave the Converter to do everything.

This felt like an overly aggressive dependency jump, so in the
interest of continuing to support Connect 3.x versions, I've
implemented a fall-back implementation that reuses the schema
"envelope" approach present in JsonConverter 3.x
The additional string operations this will incur for every
message will almost certainly impact performance, so I see this
as a temporary workaround that we should remove as soon as we
feel that Connect 4.x adoption is sufficient.

Signed-off-by: Dale Lane <[email protected]>

Author: dalelane

src/main/java/com/ibm/eventstreams/connect/mqsource/MQSourceConnector.java

@@ -33,12 +33,19 @@
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.config.ConfigValue;
 import org.apache.kafka.connect.connector.Task;
+import org.apache.kafka.connect.errors.DataException;
+import org.apache.kafka.connect.json.JsonConverter;
+import org.apache.kafka.connect.json.JsonConverterConfig;
+import org.apache.kafka.connect.json.JsonDeserializer;
 import org.apache.kafka.connect.source.ConnectorTransactionBoundaries;
 import org.apache.kafka.connect.source.ExactlyOnceSupport;
 import org.apache.kafka.connect.source.SourceConnector;
+import org.apache.kafka.connect.storage.ConverterType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.fasterxml.jackson.databind.JsonNode;
+
 public class MQSourceConnector extends SourceConnector {
     private static final Logger log = LoggerFactory.getLogger(MQSourceConnector.class);
 
@@ -212,6 +219,15 @@ public class MQSourceConnector extends SourceConnector {
             "keys, all error context header keys will start with <code>__connect.errors.</code>";
     private static final String DLQ_CONTEXT_HEADERS_ENABLE_DISPLAY = "Enable Error Context Headers";
 
+    public static final String CONFIG_NAME_MQ_RECORD_BUILDER_JSON_SCHEMAS_ENABLE = "mq.record.builder.json.schemas.enable";
+    public static final String CONFIG_DOCUMENTATION_MQ_RECORD_BUILDER_JSON_SCHEMAS_ENABLE = "Include schemas within the Kafka messages produced by the JSON record builder.";
+    public static final String CONFIG_DISPLAY_MQ_RECORD_BUILDER_JSON_SCHEMAS_ENABLE = "Enable Schemas";
+
+    public static final String CONFIG_NAME_MQ_RECORD_BUILDER_JSON_SCHEMA_CONTENT = "mq.record.builder.json.schema.content";
+    public static final String CONFIG_DOCUMENTATION_MQ_RECORD_BUILDER_JSON_SCHEMA_CONTENT = "When set, this is used as the schema for all messages. This should be a Kafka Connect schema, as used by JsonConverter.";
+    public static final String CONFIG_DISPLAY_MQ_RECORD_BUILDER_JSON_SCHEMA_CONTENT = "Schema Content";
+
+
     // Define valid reconnect options
     public static final String[] CONFIG_VALUE_MQ_VALID_RECONNECT_OPTIONS = {
         CONFIG_VALUE_MQ_CLIENT_RECONNECT_OPTION_ASDEF,
@@ -666,6 +682,22 @@ null, new ReadableFile(),
                 32,
                 ConfigDef.Width.MEDIUM,
                 CONFIG_DISPLAY_MAX_POLL_TIME);
+        CONFIGDEF.define(CONFIG_NAME_MQ_RECORD_BUILDER_JSON_SCHEMAS_ENABLE,
+                Type.BOOLEAN,
+                false, new ConfigDef.NonNullValidator(),
+                Importance.LOW,
+                CONFIG_DOCUMENTATION_MQ_RECORD_BUILDER_JSON_SCHEMAS_ENABLE,
+                CONFIG_GROUP_MQ, 33,
+                Width.SHORT,
+                CONFIG_DISPLAY_MQ_RECORD_BUILDER_JSON_SCHEMAS_ENABLE);
+        CONFIGDEF.define(CONFIG_NAME_MQ_RECORD_BUILDER_JSON_SCHEMA_CONTENT,
+                Type.STRING,
+                null, new SchemaValidator(),
+                Importance.LOW,
+                CONFIG_DOCUMENTATION_MQ_RECORD_BUILDER_JSON_SCHEMA_CONTENT,
+                CONFIG_GROUP_MQ, 34,
+                Width.MEDIUM,
+                CONFIG_DISPLAY_MQ_RECORD_BUILDER_JSON_SCHEMA_CONTENT);
 
         CONFIGDEF.define(CONFIG_NAME_TOPIC,
                 Type.STRING,
@@ -716,6 +748,43 @@ public void ensureValid(final String name, final Object value) {
         }
     }
 
+    private static class SchemaValidator implements ConfigDef.Validator {
+        @Override
+        public void ensureValid(final String name, final Object value) {
+            final String strValue = (String) value;
+            if (value == null || strValue.trim().isEmpty()) {
+                // only validate non-empty schemas
+                return;
+            }
+
+            // Start with a quick and simple "sniff test" on the provided schema
+            //  by checking if it starts and ends in curly-parentheses
+            // This will quickly catch obvious configuration errors, such as
+            //  providing a name, id, or file location for a schema
+            final String trimmedStr = strValue.trim();
+            if (!trimmedStr.startsWith("{") || !trimmedStr.endsWith("}")) {
+                throw new ConfigException(name, value, "Value should be a Kafka Connect schema");
+            }
+
+            // Create a temporary JsonDeserializer/JsonConverter to parse the
+            //  provided schema.
+            // The aim for doing this is to catch any invalid schemas at
+            //  startup time, rather than allow this to go unnoticed until
+            //  the first MQ message is received (potentially a long time
+            //  later).
+            try (
+                final JsonDeserializer deserializer = new JsonDeserializer();
+                final JsonConverter conv = new JsonConverter()
+            ) {
+                final JsonNode jsonStr = deserializer.deserialize(trimmedStr, trimmedStr.getBytes());
+                conv.configure(Map.of(JsonConverterConfig.TYPE_CONFIG, ConverterType.VALUE.getName()));
+                conv.asConnectSchema(jsonStr);
+            } catch (final DataException exc) {
+                throw new ConfigException(name, value, exc.getMessage());
+            }
+        }
+    }
+
     /**
      * Signals that this connector is not capable of defining other transaction boundaries.
      * A new transaction will be started and committed for every batch of records returned by {@link MQSourceTask#poll()}.

src/main/java/com/ibm/eventstreams/connect/mqsource/builders/JsonRecordBuilder.java

@@ -18,17 +18,23 @@
 import static java.nio.charset.StandardCharsets.UTF_8;
 
 import java.util.HashMap;
+import java.util.Map;
+
 import javax.jms.BytesMessage;
 import javax.jms.JMSContext;
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.TextMessage;
 
+import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.connect.data.SchemaAndValue;
 import org.apache.kafka.connect.json.JsonConverter;
+import org.apache.kafka.connect.json.JsonConverterConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.ibm.eventstreams.connect.mqsource.MQSourceConnector;
+
 /**
  * Builds Kafka Connect SourceRecords from messages. It parses the bytes of the payload of JMS
  * BytesMessage and TextMessage as JSON and creates a SourceRecord with a null schema.
@@ -38,13 +44,53 @@ public class JsonRecordBuilder extends BaseRecordBuilder {
 
     private JsonConverter converter;
 
+    // From Kafka Connect 4.2 onwards, JsonConverter includes schema support
+    //  To support earlier versions of the dependency, the record builder includes a
+    //  workaround implementation.
+    //  This variable should be true where the workaround implementation is required.
+    private boolean recordBuilderSchemaSupport = false;
+
+    // Workaround for supporting schemas is to embed the schema in the message payload
+    //  given to the JsonConverter. This variable contains a String to concatenate with
+    //  the string received from MQ in order to achieve this.
+    private String schemaSupportEnvelope = null;
+
     public JsonRecordBuilder() {
         log.info("Building records using com.ibm.eventstreams.connect.mqsource.builders.JsonRecordBuilder");
         converter = new JsonConverter();
+    }
+
+    /**
+     * Configure this class. In addition to the MQ message handling config
+     *  used by BaseRecordBuilder, this also configures the JsonConverter
+     *  used by this record builder to parse JSON messages from MQ.
+     */
+    @Override
+    public void configure(final Map<String, String> props) {
+        super.configure(props);
+
+        final AbstractConfig config = new AbstractConfig(MQSourceConnector.CONFIGDEF, props);
+        final boolean schemasEnable = config.getBoolean(MQSourceConnector.CONFIG_NAME_MQ_RECORD_BUILDER_JSON_SCHEMAS_ENABLE);
+        String schemaContent = null;
+        if (schemasEnable) {
+            schemaContent = config.getString(MQSourceConnector.CONFIG_NAME_MQ_RECORD_BUILDER_JSON_SCHEMA_CONTENT);
+            if (schemaContent != null) {
+                schemaContent = schemaContent.trim();
+            }
+        }
+
+        if (schemasEnable && schemaContent != null &&
+            !JsonConverterConfig.configDef().names().contains("schema.content")) {
 
-        // We just want the payload, not the schema in the output message
-        final HashMap<String, String> m = new HashMap<>();
-        m.put("schemas.enable", "false");
+            // support for schemas provided separately from message payloads is requested
+            //  but not available natively within the JsonConverter present in the classpath
+            recordBuilderSchemaSupport = true;
+            schemaSupportEnvelope = "{\"schema\": " + schemaContent + ", \"payload\": ";
+        }
+
+        final Map<String, String> m = new HashMap<>();
+        m.put("schemas.enable", Boolean.toString(schemasEnable));
+        m.put("schema.content", schemaContent);
 
         // Convert the value, not the key (isKey == false)
         converter.configure(m, false);
@@ -77,6 +123,14 @@ public SchemaAndValue getValue(final JMSContext context, final String topic, fin
             throw new RecordBuilderException("Unsupported JMS message type");
         }
 
-        return converter.toConnectData(topic, payload);
+        if (recordBuilderSchemaSupport) {
+            return converter.toConnectData(topic,
+                // embed schema in the event payload
+                (schemaSupportEnvelope + new String(payload) + "}").getBytes());
+        } else {
+            return converter.toConnectData(topic,
+                // submit the payload as-is to the converter
+                payload);
+        }
     }
 }