fix(filters): XmlToJson should support bytes input

Author: fhussonnois

connect-file-pulse-filters/src/main/java/io/streamthoughts/kafka/connect/filepulse/config/XmlToJsonFilterConfig.java

@@ -20,7 +20,10 @@
 
 import org.apache.kafka.common.config.ConfigDef;
 
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
+import java.util.Optional;
 
 public class XmlToJsonFilterConfig extends CommonFilterConfig {
 
@@ -36,6 +39,9 @@ public class XmlToJsonFilterConfig extends CommonFilterConfig {
     public static final String XML_PARSER_CDATA_TAG_NAME_DOC = "The name of the key in a JSON Object that indicates " +
                                                                "a CDATA section (default: '" + XML_PARSER_CDATA_TAG_NAME_DEFAULT + "').";
 
+    public static final String XML_PARSER_SOURCE_CHARSET_CONFIG = "source.charset";
+    public static final String XML_PARSER_SOURCE_CHARSET_DOC = "The charset to be used for reading the source " +
+            "                                                   field (default: UTF-8)";
 
     /**
      * Creates a new {@link XmlToJsonFilterConfig} instance.
@@ -72,6 +78,17 @@ public static ConfigDef configDef() {
                         filterGroupCounter++,
                         ConfigDef.Width.NONE,
                         XML_PARSER_CDATA_TAG_NAME_CONFIG
+                )
+                .define(
+                        XML_PARSER_SOURCE_CHARSET_CONFIG,
+                        ConfigDef.Type.STRING,
+                        null,
+                        ConfigDef.Importance.MEDIUM,
+                        XML_PARSER_SOURCE_CHARSET_DOC,
+                        FILTER,
+                        filterGroupCounter++,
+                        ConfigDef.Width.NONE,
+                        XML_PARSER_SOURCE_CHARSET_CONFIG
                 );
     }
 
@@ -82,4 +99,10 @@ public boolean getXmlParserKeepStrings() {
     public String getCDataTagName() {
         return getString(XML_PARSER_CDATA_TAG_NAME_CONFIG);
     }
+
+    public Charset charset() {
+        return Optional.ofNullable(getString(XML_PARSER_SOURCE_CHARSET_CONFIG))
+                .map(Charset::forName)
+                .orElse(StandardCharsets.UTF_8);
+    }
 }

connect-file-pulse-filters/src/main/java/io/streamthoughts/kafka/connect/filepulse/filter/XmlToJsonFilter.java

@@ -29,15 +29,20 @@
 import org.json.XML;
 import org.json.XMLParserConfiguration;
 
+import java.io.ByteArrayInputStream;
+import java.io.InputStreamReader;
+import java.io.Reader;
+import java.io.StringReader;
 import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
 
-public class XmlToJsonFilter extends AbstractMergeRecordFilter<XmlToJsonFilter>  {
+public class XmlToJsonFilter extends AbstractMergeRecordFilter<XmlToJsonFilter> {
 
     private XmlToJsonFilterConfig config;
 
     private XMLParserConfiguration xmlParserConfiguration;
+
     /**
      * {@inheritDoc}
      */
@@ -55,8 +60,8 @@ public void configure(final Map<String, ?> props) {
         config = new XmlToJsonFilterConfig(props);
 
         xmlParserConfiguration = new XMLParserConfiguration()
-            .withKeepStrings(config.getXmlParserKeepStrings())
-            .withcDataTagName(config.getCDataTagName());
+                .withKeepStrings(config.getXmlParserKeepStrings())
+                .withcDataTagName(config.getCDataTagName());
     }
 
     /**
@@ -65,14 +70,33 @@ public void configure(final Map<String, ?> props) {
     @Override
     protected RecordsIterable<TypedStruct> apply(final FilterContext context,
                                                  final TypedStruct record) throws FilterException {
-        try {
-            final String payload = checkIsNotNull(record.get(config.source())).getString();
 
-            if (StringUtils.isBlank(payload)) {
-                return RecordsIterable.empty();
-            }
+        final TypedValue value = checkIsNotNull(record.get(config.source()));
+        switch (value.type()) {
+            case STRING:
+                final String xml = value.getString();
+                if (StringUtils.isBlank(value.getString())) {
+                    return RecordsIterable.empty();
+                }
+                return xmlToJson(new StringReader(xml));
+            case BYTES:
+                byte[] bytes = value.getBytes();
+                if (bytes.length == 0) {
+                    return RecordsIterable.empty();
+                }
+                return xmlToJson(new InputStreamReader(new ByteArrayInputStream(bytes), config.charset()));
+            default:
+                throw new FilterException(
+                        "Invalid field '" + config.source() + "' was passed through the " +
+                        "connector's configuration'. " +
+                        "Cannot parse field of type '" + value.type() + "' to XML."
+                );
+        }
+    }
 
-            final JSONObject xmlJSONObj = XML.toJSONObject(payload, xmlParserConfiguration);
+    private RecordsIterable<TypedStruct> xmlToJson(final Reader xml) {
+        try {
+            final JSONObject xmlJSONObj = XML.toJSONObject(xml, xmlParserConfiguration);
             final String jsonString = xmlJSONObj.toString(0);
             return RecordsIterable.of(TypedStruct.create().put(config.source(), jsonString));
         } catch (JSONException e) {
@@ -92,7 +116,7 @@ private TypedValue checkIsNotNull(final TypedValue value) {
         if (value.isNull()) {
             throw new FilterException(
                     "Invalid field '" + config.source() + "' was passed through the connector's configuration'. " +
-                    "Cannot parse null or empty value to XML."
+                            "Cannot parse null or empty value to XML."
             );
         }
         return value;

connect-file-pulse-filters/src/test/java/io/streamthoughts/kafka/connect/filepulse/filter/XmlToJsonFilterTest.java

@@ -22,6 +22,7 @@
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 
 public class XmlToJsonFilterTest {
@@ -37,7 +38,7 @@ public class XmlToJsonFilterTest {
             "</root>";
 
     @Test
-    public void should_success_to_convert_xml_to_json() {
+    public void should_success_to_convert_xml_to_json_given_input_string() {
         final XmlToJsonFilter filter = new XmlToJsonFilter();
         filter.configure(Collections.emptyMap());
         final TypedStruct input = TypedStruct.create().put("message", XML);
@@ -49,4 +50,18 @@ public void should_success_to_convert_xml_to_json() {
             output.getString("message")
         );
     }
+
+    @Test
+    public void should_success_to_convert_xml_to_json_given_input_bytes() {
+        final XmlToJsonFilter filter = new XmlToJsonFilter();
+        filter.configure(Collections.emptyMap());
+        final TypedStruct input = TypedStruct.create().put("message", XML.getBytes(StandardCharsets.UTF_8));
+        final TypedStruct output = filter.apply(null, input).iterator().next();
+
+        Assert.assertNotNull(output);
+        Assert.assertEquals(
+                "{\"root\":{\"element1\":{\"attr\":\"foo\",\"value\":\"bar\"},\"element2\":\"value\",\"element3\":\"value\",\"element4\":[\"value1\",\"value2\"],\"element5\":\"\"}}",
+                output.getString("message")
+        );
+    }
 }