feat(dataformat): add new config props to exclude all XML attributes (#174)

Resolves: #174

Author: fhussonnois

connect-file-pulse-dataformat/src/main/java/io/streamthoughts/kafka/connect/filepulse/xml/XMLCommonConfig.java

@@ -43,6 +43,9 @@ public class XMLCommonConfig extends AbstractConfig {
     public static final String XML_EXCLUDE_EMPTY_ELEMENTS_CONFIG = "xml.exclude.empty.elements";
     private static final String XML_EXCLUDE_EMPTY_ELEMENTS_DOC = "Specifies that the reader should exclude element having no field (default: false).";
 
+    public static final String XML_EXCLUDE_NODE_ATTRIBUTES_CONFIG = "xml.exclude.node.attributes";
+    private static final String XML_EXCLUDE_NODE_ATTRIBUTES_DOC = "Specifies that the reader should exclude node attributes (default: false).";
+
     public static final String XML_DATA_TYPE_INFERENCE_ENABLED_CONFIG = "xml.data.type.inference.enabled";
     private static final String XML_DATA_TYPE_INFERENCE_ENABLED_DOC = "Specifies that the reader should try to infer the type of data nodes. (default: false).";
 
@@ -76,6 +79,10 @@ public boolean isEmptyElementExcluded() {
         return getBoolean(withKeyPrefix(XML_EXCLUDE_EMPTY_ELEMENTS_CONFIG));
     }
 
+    public boolean isNodeAttributesExcluded() {
+        return getBoolean(withKeyPrefix(XML_EXCLUDE_NODE_ATTRIBUTES_CONFIG));
+    }
+
     public boolean isDataTypeInferenceEnabled() {
         return getBoolean(withKeyPrefix(XML_DATA_TYPE_INFERENCE_ENABLED_CONFIG));
     }
@@ -138,6 +145,17 @@ public static ConfigDef buildConfigDefWith(final String group,
                         ConfigDef.Width.NONE,
                         keyPrefix + XML_EXCLUDE_EMPTY_ELEMENTS_CONFIG
                 )
+                .define(
+                        keyPrefix + XML_EXCLUDE_NODE_ATTRIBUTES_CONFIG,
+                        ConfigDef.Type.BOOLEAN,
+                        false,
+                        ConfigDef.Importance.LOW,
+                        XML_EXCLUDE_NODE_ATTRIBUTES_DOC,
+                        group,
+                        filterGroupCounter++,
+                        ConfigDef.Width.NONE,
+                        keyPrefix + XML_EXCLUDE_NODE_ATTRIBUTES_CONFIG
+                )
                 .define(
                         keyPrefix + XML_DATA_TYPE_INFERENCE_ENABLED_CONFIG,
                         ConfigDef.Type.BOOLEAN,

connect-file-pulse-dataformat/src/main/java/io/streamthoughts/kafka/connect/filepulse/xml/XMLNodeToStructConverter.java

@@ -58,6 +58,8 @@ public final class XMLNodeToStructConverter implements Function<Node, TypedStruc
 
     private boolean excludeEmptyElement = false;
 
+    private boolean excludeNodeAttributes = false;
+
     private boolean isTypeInferenceEnabled = false;
 
     private FieldPaths forceArrayFields = FieldPaths.empty();
@@ -67,6 +69,12 @@ public XMLNodeToStructConverter setExcludeEmptyElement(boolean excludeEmptyEleme
         return this;
     }
 
+    public XMLNodeToStructConverter setExcludeNodeAttributes(boolean excludeNodeAttributes) {
+        this.excludeNodeAttributes = excludeNodeAttributes;
+        return this;
+    }
+
+
     public XMLNodeToStructConverter setForceArrayFields(final FieldPaths forceArrayFields) {
         this.forceArrayFields = forceArrayFields;
         return this;
@@ -149,7 +157,7 @@ private TypedValue toTypedValue(final String text) {
         return isTypeInferenceEnabled ? TypedValue.parse(text) : TypedValue.string(text);
     }
 
-    private static Optional<TypedValue> readTextNodeValue(final Node node, final TypedValue data) {
+    private Optional<TypedValue> readTextNodeValue(final Node node, final TypedValue data) {
         // Check if TextNode as no attribute
         final NamedNodeMap attributes = node.getAttributes();
         if (attributes == null || attributes.getLength() == 0) {
@@ -240,7 +248,7 @@ private static List<Node> collectAllNotNewLineNodes(final NodeList nodes) {
 
     private static boolean isTextNode(final Node n) {
         return isNodeOfType(n, Node.TEXT_NODE) ||
-                isNodeOfType(n, Node.CDATA_SECTION_NODE);
+               isNodeOfType(n, Node.CDATA_SECTION_NODE);
     }
 
     private static boolean isElementNode(final Node n) {
@@ -251,9 +259,9 @@ private static boolean isNodeOfType(final Node node, int textNode) {
         return node.getNodeType() == textNode;
     }
 
-    private static void addAllNodeAttributes(final TypedStruct struct,
-                                             final NamedNodeMap attributes) {
-        if (attributes == null) return;
+    private void addAllNodeAttributes(final TypedStruct struct,
+                                      final NamedNodeMap attributes) {
+        if (excludeNodeAttributes || attributes == null) return;
 
         for (int i = 0; i < attributes.getLength(); i++) {
             Attr attr = (Attr) attributes.item(i);

connect-file-pulse-filesystems/filepulse-commons-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/xml/XMLFileInputIterator.java

@@ -82,6 +82,7 @@ public XMLFileInputIterator(final XMLFileInputReaderConfig config,
 
         this.converter = new XMLNodeToStructConverter()
                 .setExcludeEmptyElement(config.isEmptyElementExcluded())
+                .setExcludeNodeAttributes(config.isNodeAttributesExcluded())
                 .setForceArrayFields(FieldPaths.from(config.forceArrayFields()))
                 .setTypeInferenceEnabled(config.isDataTypeInferenceEnabled());
 

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

@@ -72,6 +72,7 @@ public void configure(final Map<String, ?> configs) {
         var filterConfig = new XmlToStructFilterConfig(configs);
         this.converter = new XMLNodeToStructConverter()
             .setExcludeEmptyElement(filterConfig.isEmptyElementExcluded())
+            .setExcludeNodeAttributes(filterConfig.isNodeAttributesExcluded())
             .setForceArrayFields(FieldPaths.from(filterConfig.forceArrayFields()))
             .setTypeInferenceEnabled(filterConfig.isDataTypeInferenceEnabled());