fix(dataformat): fix invalig config and NPE in XMLNodeToStructConverter

Author: fhussonnois

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

@@ -113,7 +113,7 @@ public String getAttributePrefix() {
         return getString(withKeyPrefix(XML_ATTRIBUTE_PREFIX_CONFIG));
     }
 
-    public String getTextNodeValueFieldName() {
+    public String getContentFieldName() {
         return getString(withKeyPrefix(XML_CONTENT_FIELD_NAME_CONFIG));
     }
 
@@ -126,7 +126,7 @@ public String getXmlFieldCharactersStringReplacement() {
     }
 
     public List<String> getForceContentFields() {
-        return getList(withKeyPrefix(XML_CONTENT_FIELD_NAME_CONFIG));
+        return getList(withKeyPrefix(XML_FORCE_CONTENT_FIELD_FOR_PATHS_CONFIG));
     }
 
     public boolean isDataTypeInferenceEnabled() {

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

@@ -165,6 +165,7 @@ private TypedValue convertObjectTree(final Node node,
             final String childNodeName = isTextNode(child) ? nodeName : determineNodeName(child);
             Optional<TypedValue> optional = readObjectNodeValue(
                     child,
+                    childNodeName,
                     currentForceArrayFields,
                     currentForceContentFields
             );
@@ -184,20 +185,19 @@ private TypedValue convertObjectTree(final Node node,
     }
 
     private Optional<TypedValue> readObjectNodeValue(final Node node,
+                                                     final String nodeName,
                                                      final FieldPaths forceArrayFields,
                                                      final FieldPaths forceContentFields) {
         if (isWhitespaceOrNewLineNodeElement(node) ||
             isNodeOfType(node, Node.DOCUMENT_TYPE_NODE)) {
             return Optional.empty();
         }
 
-        final boolean forceContentField = forceContentFields.anyMatches(determineNodeName(node));
-
         if (isTextNode(node)) {
             final String text = node.getNodeValue();
             final TypedValue data = toTypedValue(text);
 
-            return readTextNodeValue(node, data, forceContentField);
+            return readTextNodeValue(node, data, forceContentFields.anyMatches(nodeName));
         }
 
         if (isElementNode(node)) {
@@ -206,7 +206,7 @@ private Optional<TypedValue> readObjectNodeValue(final Node node,
             if (childTextContent.isPresent()) {
                 final String text = childTextContent.get();
                 final TypedValue data = toTypedValue(text);
-                return readTextNodeValue(node, data, forceContentField);
+                return readTextNodeValue(node, data, forceContentFields.anyMatches(nodeName));
             }
             return Optional.of(convertObjectTree(node, forceArrayFields, forceContentFields));
         }
@@ -243,7 +243,8 @@ private Map<String, String> getNotExcludedNodeAttributes(final Node node) {
         final Map<String, String> attributes = new HashMap<>();
         for (int i = 0; i < nodeMap.getLength(); i++) {
             Attr attr = (Attr) nodeMap.item(i);
-            if (!excludeAttributesInNamespaces.contains(attr.getNamespaceURI())) {
+            String namespaceURI = attr.getNamespaceURI();
+            if (namespaceURI == null || !excludeAttributesInNamespaces.contains(namespaceURI)) {
                 String attrName = determineNodeName(attr);
                 if (isNotXmlNamespace(attr)) {
                     attributes.put(attributePrefix + attrName, attr.getNodeValue());

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

@@ -89,7 +89,7 @@ public XMLFileInputIterator(final XMLFileInputReaderConfig config,
                 .setForceArrayFields(FieldPaths.from(config.forceArrayFields()))
                 .setForceContentFields(FieldPaths.from(config.getForceContentFields()))
                 .setTypeInferenceEnabled(config.isDataTypeInferenceEnabled())
-                .setContentFieldName(config.getTextNodeValueFieldName())
+                .setContentFieldName(config.getContentFieldName())
                 .setFieldCharactersRegexPattern(config.getXmlFieldCharactersRegexPattern())
                 .setFieldCharactersStringReplacement(config.getXmlFieldCharactersStringReplacement())
                 .setAttributePrefix(config.getAttributePrefix());

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

@@ -77,7 +77,7 @@ public void configure(final Map<String, ?> configs) {
                 .setForceArrayFields(FieldPaths.from(filterConfig.forceArrayFields()))
                 .setForceContentFields(FieldPaths.from(filterConfig.getForceContentFields()))
                 .setTypeInferenceEnabled(filterConfig.isDataTypeInferenceEnabled())
-                .setContentFieldName(filterConfig.getTextNodeValueFieldName())
+                .setContentFieldName(filterConfig.getContentFieldName())
                 .setFieldCharactersRegexPattern(filterConfig.getXmlFieldCharactersRegexPattern())
                 .setFieldCharactersStringReplacement(filterConfig.getXmlFieldCharactersStringReplacement())
                 .setAttributePrefix(filterConfig.getAttributePrefix());