feat(filters): add new simple XmlToJsonFilter

Author: fhussonnois

connect-file-pulse-filters/pom.xml

@@ -81,6 +81,12 @@
             <version>1.1.0</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.json</groupId>
+            <artifactId>json</artifactId>
+            <version>20210307</version>
+        </dependency>
+
         <dependency>
             <groupId>com.jsoniter</groupId>
             <artifactId>jsoniter</artifactId>

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

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2021 StreamThoughts.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamthoughts.kafka.connect.filepulse.config;
+
+import org.apache.kafka.common.config.ConfigDef;
+
+import java.util.Map;
+
+public class XmlToJsonFilterConfig extends CommonFilterConfig {
+
+    private static final String FILTER = "XmlToJsonFilter";
+
+    public static final String XML_PARSER_KEEP_STRINGS_CONFIG = "xml.parser.keep.strings";
+    public static final String XML_PARSER_KEEP_STRINGS_DOC = "When parsing the XML into JSON, specifies if values should be kept " +
+                                                         "as strings (true), or if " +
+                                                         "they should try to be guessed into JSON values (numeric, boolean, string)";
+
+    public static final String XML_PARSER_CDATA_TAG_NAME_DEFAULT = "value";
+    public static final String XML_PARSER_CDATA_TAG_NAME_CONFIG = "xml.parser.cDataTagName";
+    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 + "').";
+
+
+    /**
+     * Creates a new {@link XmlToJsonFilterConfig} instance.
+     *
+     * @param originals the originals configuration.
+     */
+    public XmlToJsonFilterConfig(final Map<?, ?> originals) {
+        super(configDef(), originals);
+    }
+
+    public static ConfigDef configDef() {
+        int filterGroupCounter = 0;
+        return new ConfigDef(CommonFilterConfig.configDef())
+                .define(withOverwrite(FILTER, filterGroupCounter++))
+                .define(withSource(FILTER, filterGroupCounter++))
+                .define(
+                        XML_PARSER_KEEP_STRINGS_CONFIG,
+                        ConfigDef.Type.BOOLEAN,
+                        false,
+                        ConfigDef.Importance.HIGH,
+                        XML_PARSER_KEEP_STRINGS_DOC,
+                        FILTER,
+                        filterGroupCounter++,
+                        ConfigDef.Width.NONE,
+                        XML_PARSER_KEEP_STRINGS_CONFIG
+                )
+                .define(
+                        XML_PARSER_CDATA_TAG_NAME_CONFIG,
+                        ConfigDef.Type.STRING,
+                        XML_PARSER_CDATA_TAG_NAME_DEFAULT,
+                        ConfigDef.Importance.MEDIUM,
+                        XML_PARSER_CDATA_TAG_NAME_DOC,
+                        FILTER,
+                        filterGroupCounter++,
+                        ConfigDef.Width.NONE,
+                        XML_PARSER_CDATA_TAG_NAME_CONFIG
+                );
+    }
+
+    public boolean getXmlParserKeepStrings() {
+        return getBoolean(XML_PARSER_KEEP_STRINGS_CONFIG);
+    }
+
+    public String getCDataTagName() {
+        return getString(XML_PARSER_CDATA_TAG_NAME_CONFIG);
+    }
+}

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

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2021 StreamThoughts.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamthoughts.kafka.connect.filepulse.filter;
+
+import io.streamthoughts.kafka.connect.filepulse.config.XmlToJsonFilterConfig;
+import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
+import io.streamthoughts.kafka.connect.filepulse.data.TypedValue;
+import io.streamthoughts.kafka.connect.filepulse.internal.StringUtils;
+import io.streamthoughts.kafka.connect.filepulse.reader.RecordsIterable;
+import org.apache.kafka.common.config.ConfigDef;
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.json.XML;
+import org.json.XMLParserConfiguration;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
+public class XmlToJsonFilter extends AbstractMergeRecordFilter<XmlToJsonFilter>  {
+
+    private XmlToJsonFilterConfig config;
+
+    private XMLParserConfiguration xmlParserConfiguration;
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public ConfigDef configDef() {
+        return XmlToJsonFilterConfig.configDef();
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public void configure(final Map<String, ?> props) {
+        super.configure(props);
+        config = new XmlToJsonFilterConfig(props);
+
+        xmlParserConfiguration = new XMLParserConfiguration()
+            .withKeepStrings(config.getXmlParserKeepStrings())
+            .withcDataTagName(config.getCDataTagName());
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @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 JSONObject xmlJSONObj = XML.toJSONObject(payload, xmlParserConfiguration);
+            final String jsonString = xmlJSONObj.toString(0);
+            return RecordsIterable.of(TypedStruct.create().put(config.source(), jsonString));
+        } catch (JSONException e) {
+            throw new FilterException("Failed to parse and convert XML document into JSON object", e);
+        }
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    protected Set<String> overwrite() {
+        return Collections.singleton(config.source());
+    }
+
+    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."
+            );
+        }
+        return value;
+    }
+}

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

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2021 StreamThoughts.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamthoughts.kafka.connect.filepulse.filter;
+
+import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+
+public class XmlToJsonFilterTest {
+
+    String XML = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n" +
+            "<root>\n" +
+            "  <element1 attr=\"foo\">bar</element1>\n" +
+            "  <element2>value</element2>\n" +
+            "  <element3>value</element3>\n" +
+            "  <element4>value1</element4>\n" +
+            "  <element4>value2</element4>\n" +
+            "  <element5></element5>\n" +
+            "</root>";
+
+    @Test
+    public void should_success_to_convert_xml_to_json() {
+        final XmlToJsonFilter filter = new XmlToJsonFilter();
+        filter.configure(Collections.emptyMap());
+        final TypedStruct input = TypedStruct.create().put("message", XML);
+        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")
+        );
+    }
+}