feat(dataformat): add config to specify a prefix used to prepend XML attributes (#176)

fhussonnois · fhussonnois · commit 7fea775dd3a9 · 2021-09-26T18:55:35.000+02:00
Resolves: #176
diff --git a/connect-file-pulse-dataformat/pom.xml b/connect-file-pulse-dataformat/pom.xml
@@ -30,5 +30,11 @@
             <groupId>org.apache.kafka</groupId>
             <artifactId>connect-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>${junit.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>
diff --git a/connect-file-pulse-dataformat/src/main/java/io/streamthoughts/kafka/connect/filepulse/xml/XMLCommonConfig.java b/connect-file-pulse-dataformat/src/main/java/io/streamthoughts/kafka/connect/filepulse/xml/XMLCommonConfig.java
@@ -52,7 +52,10 @@ public class XMLCommonConfig extends AbstractConfig {
     private static final String XML_EXCLUDE_NODE_ATTRIBUTES_IN_NAMESPACES_DOC = "Specifies that the reader should only exclude node attributes in the defined list of namespaces.";
 
     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).";
+    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).";
+
+    public static final String XML_ATTRIBUTE_PREFIX_CONFIG = "xml.attribute.prefix";
+    public static final String XML_ATTRIBUTE_PREFIX_DOC = "If set, the name of attributes will be prepended with the specified prefix when they are added to a record (default: '').";
 
     private final String keyPrefix;
 
@@ -92,6 +95,10 @@ public Set<String> getExcludeNodeAttributesInNamespaces() {
         return new HashSet<>(getList(withKeyPrefix(XML_EXCLUDE_NODE_ATTRIBUTES_IN_NAMESPACES_CONFIG)));
     }
 
+    public String getAttributePrefix() {
+        return getString(withKeyPrefix(XML_ATTRIBUTE_PREFIX_CONFIG));
+    }
+
     public boolean isDataTypeInferenceEnabled() {
         return getBoolean(withKeyPrefix(XML_DATA_TYPE_INFERENCE_ENABLED_CONFIG));
     }
@@ -186,6 +193,17 @@ public static ConfigDef buildConfigDefWith(final String group,
                         filterGroupCounter++,
                         ConfigDef.Width.NONE,
                         keyPrefix + XML_EXCLUDE_EMPTY_ELEMENTS_CONFIG
+                )
+                .define(
+                        keyPrefix + XML_ATTRIBUTE_PREFIX_CONFIG,
+                        ConfigDef.Type.STRING,
+                        "",
+                        ConfigDef.Importance.LOW,
+                        XML_ATTRIBUTE_PREFIX_DOC,
+                        group,
+                        filterGroupCounter++,
+                        ConfigDef.Width.NONE,
+                        keyPrefix + XML_ATTRIBUTE_PREFIX_CONFIG
                 );
 
         for (ConfigDef.ConfigKey configKey : additional) {
diff --git a/connect-file-pulse-dataformat/src/main/java/io/streamthoughts/kafka/connect/filepulse/xml/XMLNodeToStructConverter.java b/connect-file-pulse-dataformat/src/main/java/io/streamthoughts/kafka/connect/filepulse/xml/XMLNodeToStructConverter.java
@@ -68,6 +68,8 @@ public final class XMLNodeToStructConverter implements Function<Node, TypedStruc
 
     private Set<String> excludeAttributesInNamespaces = Collections.emptySet();
 
+    private String attributePrefix = "";
+
     private FieldPaths forceArrayFields = FieldPaths.empty();
 
     public XMLNodeToStructConverter setExcludeEmptyElement(boolean excludeEmptyElement) {
@@ -85,6 +87,11 @@ public XMLNodeToStructConverter setExcludeAttributesInNamespaces(final Set<Strin
         return this;
     }
 
+    public XMLNodeToStructConverter setAttributePrefix(final String attributePrefix) {
+        this.attributePrefix = attributePrefix;
+        return this;
+    }
+
     public XMLNodeToStructConverter setForceArrayFields(final FieldPaths forceArrayFields) {
         this.forceArrayFields = forceArrayFields;
         return this;
@@ -193,7 +200,7 @@ private Map<String, String> getNotExcludedNodeAttributes(final Node node) {
             if (!excludeAttributesInNamespaces.contains(attr.getNamespaceURI())) {
                 String attrName = determineNodeName(attr);
                 if (isNotXmlNamespace(attr)) {
-                    attributes.put(attrName, attr.getNodeValue());
+                    attributes.put(attributePrefix + attrName, attr.getNodeValue());
                 }
             }
         }
diff --git a/connect-file-pulse-dataformat/src/test/java/io/streamthoughts/kafka/connect/filepulse/xml/XMLNodeToStructConverterTest.java b/connect-file-pulse-dataformat/src/test/java/io/streamthoughts/kafka/connect/filepulse/xml/XMLNodeToStructConverterTest.java
@@ -0,0 +1,65 @@
+/*
+ * 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.xml;
+
+import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
+import org.junit.Assert;
+import org.junit.Test;
+import org.w3c.dom.Document;
+import org.xml.sax.SAXException;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+
+public class XMLNodeToStructConverterTest {
+
+    private final XMLDocumentReader reader = new XMLDocumentReader(true, true);
+
+    @Test
+    public void should_prefix_node_attributes() throws IOException, SAXException {
+
+        // Given
+        final XMLNodeToStructConverter converter = new XMLNodeToStructConverter()
+                .setAttributePrefix("prefix_");
+
+        // When
+        final Document document = reader.parse(new ByteArrayInputStream("<root attr=\"text\">test</root>".getBytes()));
+        final TypedStruct result = converter.apply(document);
+
+        // Then
+        final String path = "root.prefix_attr";
+        Assert.assertTrue(result.exists(path));
+        Assert.assertEquals("text", result.find(path).getString());
+    }
+
+    @Test
+    public void should_not_prefix_node_attributes() throws IOException, SAXException {
+        // Given
+        final XMLNodeToStructConverter converter = new XMLNodeToStructConverter();
+
+        // When
+        final Document document = reader.parse(new ByteArrayInputStream("<root attr=\"text\">test</root>".getBytes()));
+        final TypedStruct result = converter.apply(document);
+
+        // Then
+        final String path = "root.attr";
+        Assert.assertTrue(result.exists(path));
+        Assert.assertEquals("text", result.find(path).getString());
+    }
+}
diff --git a/connect-file-pulse-filesystems/filepulse-amazons3-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/AmazonS3XMLFileInputReader.java b/connect-file-pulse-filesystems/filepulse-amazons3-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/AmazonS3XMLFileInputReader.java
@@ -19,8 +19,8 @@
 package io.streamthoughts.kafka.connect.filepulse.fs.reader;
 
 import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
-import io.streamthoughts.kafka.connect.filepulse.xml.XMLFileInputIteratorFactory;
-import io.streamthoughts.kafka.connect.filepulse.xml.XMLFileInputReaderConfig;
+import io.streamthoughts.kafka.connect.filepulse.fs.reader.xml.XMLFileInputIteratorFactory;
+import io.streamthoughts.kafka.connect.filepulse.fs.reader.xml.XMLFileInputReaderConfig;
 import io.streamthoughts.kafka.connect.filepulse.reader.FileInputIterator;
 import io.streamthoughts.kafka.connect.filepulse.source.FileRecord;
 
diff --git a/connect-file-pulse-filesystems/filepulse-azure-storage-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/AzureBlobStorageXMLFileInputReader.java b/connect-file-pulse-filesystems/filepulse-azure-storage-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/AzureBlobStorageXMLFileInputReader.java
@@ -19,8 +19,8 @@
 package io.streamthoughts.kafka.connect.filepulse.fs.reader;
 
 import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
-import io.streamthoughts.kafka.connect.filepulse.xml.XMLFileInputIteratorFactory;
-import io.streamthoughts.kafka.connect.filepulse.xml.XMLFileInputReaderConfig;
+import io.streamthoughts.kafka.connect.filepulse.fs.reader.xml.XMLFileInputIteratorFactory;
+import io.streamthoughts.kafka.connect.filepulse.fs.reader.xml.XMLFileInputReaderConfig;
 import io.streamthoughts.kafka.connect.filepulse.reader.FileInputIterator;
 import io.streamthoughts.kafka.connect.filepulse.source.FileRecord;
 
diff --git a/connect-file-pulse-filesystems/filepulse-commons-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/xml/XMLFileInputIterator.java b/connect-file-pulse-filesystems/filepulse-commons-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/xml/XMLFileInputIterator.java
@@ -16,7 +16,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamthoughts.kafka.connect.filepulse.xml;
+package io.streamthoughts.kafka.connect.filepulse.fs.reader.xml;
 
 import io.streamthoughts.kafka.connect.filepulse.data.FieldPaths;
 import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
@@ -29,6 +29,8 @@
 import io.streamthoughts.kafka.connect.filepulse.source.FileObjectOffset;
 import io.streamthoughts.kafka.connect.filepulse.source.FileRecord;
 import io.streamthoughts.kafka.connect.filepulse.source.TypedFileRecord;
+import io.streamthoughts.kafka.connect.filepulse.xml.XMLDocumentReader;
+import io.streamthoughts.kafka.connect.filepulse.xml.XMLNodeToStructConverter;
 import net.sf.saxon.lib.NamespaceConstant;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -85,12 +87,8 @@ public XMLFileInputIterator(final XMLFileInputReaderConfig config,
                 .setExcludeAllAttributes(config.isNodeAttributesExcluded())
                 .setExcludeAttributesInNamespaces(config.getExcludeNodeAttributesInNamespaces())
                 .setForceArrayFields(FieldPaths.from(config.forceArrayFields()))
-                .setTypeInferenceEnabled(config.isDataTypeInferenceEnabled());
-
-        System.setProperty(
-                "javax.xml.xpath.XPathFactory:" + NamespaceConstant.OBJECT_MODEL_SAXON,
-                "net.sf.saxon.xpath.XPathFactoryImpl"
-        );
+                .setTypeInferenceEnabled(config.isDataTypeInferenceEnabled())
+                .setAttributePrefix(config.getAttributePrefix());
 
         final QName qName = new QName("http://www.w3.org/1999/XSL/Transform", config.resultType());
 
diff --git a/connect-file-pulse-filesystems/filepulse-commons-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/xml/XMLFileInputIteratorFactory.java b/connect-file-pulse-filesystems/filepulse-commons-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/xml/XMLFileInputIteratorFactory.java
@@ -16,7 +16,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamthoughts.kafka.connect.filepulse.xml;
+package io.streamthoughts.kafka.connect.filepulse.fs.reader.xml;
 
 import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
 import io.streamthoughts.kafka.connect.filepulse.reader.FileInputIteratorFactory;
diff --git a/connect-file-pulse-filesystems/filepulse-commons-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/xml/XMLFileInputReaderConfig.java b/connect-file-pulse-filesystems/filepulse-commons-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/xml/XMLFileInputReaderConfig.java
@@ -16,8 +16,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamthoughts.kafka.connect.filepulse.xml;
+package io.streamthoughts.kafka.connect.filepulse.fs.reader.xml;
 
+import io.streamthoughts.kafka.connect.filepulse.xml.XMLCommonConfig;
 import org.apache.kafka.common.config.ConfigDef;
 
 import javax.xml.xpath.XPathConstants;
diff --git a/connect-file-pulse-filesystems/filepulse-commons-fs/src/test/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/xml/XMLFileInputIteratorTest.java b/connect-file-pulse-filesystems/filepulse-commons-fs/src/test/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/xml/XMLFileInputIteratorTest.java
@@ -16,16 +16,19 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.streamthoughts.kafka.connect.filepulse.xml;
+package io.streamthoughts.kafka.connect.filepulse.fs.reader.xml;
 
 import io.streamthoughts.kafka.connect.filepulse.data.Type;
 import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
 import io.streamthoughts.kafka.connect.filepulse.data.TypedValue;
 import io.streamthoughts.kafka.connect.filepulse.fs.reader.IteratorManager;
+import io.streamthoughts.kafka.connect.filepulse.fs.reader.xml.XMLFileInputIterator;
+import io.streamthoughts.kafka.connect.filepulse.fs.reader.xml.XMLFileInputReaderConfig;
 import io.streamthoughts.kafka.connect.filepulse.source.FileObjectMeta;
 import io.streamthoughts.kafka.connect.filepulse.source.FileRecord;
 import io.streamthoughts.kafka.connect.filepulse.source.LocalFileObjectMeta;
 import io.streamthoughts.kafka.connect.filepulse.source.TypedFileRecord;
+import io.streamthoughts.kafka.connect.filepulse.xml.XMLCommonConfig;
 import org.junit.Assert;
 import org.junit.Rule;
 import org.junit.Test;
@@ -42,7 +45,7 @@
 import java.util.HashMap;
 import java.util.List;
 
-import static io.streamthoughts.kafka.connect.filepulse.xml.XMLFileInputReaderConfig.withKeyPrefix;
+import static io.streamthoughts.kafka.connect.filepulse.fs.reader.xml.XMLFileInputReaderConfig.withKeyPrefix;
 
 public class XMLFileInputIteratorTest {
 
diff --git a/connect-file-pulse-filesystems/filepulse-google-cloud-storage-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/GcsXMLFileInputReader.java b/connect-file-pulse-filesystems/filepulse-google-cloud-storage-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/GcsXMLFileInputReader.java
@@ -19,8 +19,8 @@
 package io.streamthoughts.kafka.connect.filepulse.fs.reader;
 
 import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
-import io.streamthoughts.kafka.connect.filepulse.xml.XMLFileInputIteratorFactory;
-import io.streamthoughts.kafka.connect.filepulse.xml.XMLFileInputReaderConfig;
+import io.streamthoughts.kafka.connect.filepulse.fs.reader.xml.XMLFileInputIteratorFactory;
+import io.streamthoughts.kafka.connect.filepulse.fs.reader.xml.XMLFileInputReaderConfig;
 import io.streamthoughts.kafka.connect.filepulse.reader.FileInputIterator;
 import io.streamthoughts.kafka.connect.filepulse.source.FileRecord;
 
diff --git a/connect-file-pulse-filesystems/filepulse-local-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/LocalXMLFileInputReader.java b/connect-file-pulse-filesystems/filepulse-local-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/reader/LocalXMLFileInputReader.java
@@ -19,8 +19,8 @@
 package io.streamthoughts.kafka.connect.filepulse.fs.reader;
 
 import io.streamthoughts.kafka.connect.filepulse.data.TypedStruct;
-import io.streamthoughts.kafka.connect.filepulse.xml.XMLFileInputIteratorFactory;
-import io.streamthoughts.kafka.connect.filepulse.xml.XMLFileInputReaderConfig;
+import io.streamthoughts.kafka.connect.filepulse.fs.reader.xml.XMLFileInputIteratorFactory;
+import io.streamthoughts.kafka.connect.filepulse.fs.reader.xml.XMLFileInputReaderConfig;
 import io.streamthoughts.kafka.connect.filepulse.reader.FileInputIterator;
 import io.streamthoughts.kafka.connect.filepulse.reader.FileInputReader;
 import io.streamthoughts.kafka.connect.filepulse.source.FileRecord;
diff --git a/connect-file-pulse-filters/src/main/java/io/streamthoughts/kafka/connect/filepulse/filter/XmlToStructFilter.java b/connect-file-pulse-filters/src/main/java/io/streamthoughts/kafka/connect/filepulse/filter/XmlToStructFilter.java
@@ -75,7 +75,8 @@ public void configure(final Map<String, ?> configs) {
             .setExcludeAllAttributes(filterConfig.isNodeAttributesExcluded())
             .setExcludeAttributesInNamespaces(filterConfig.getExcludeNodeAttributesInNamespaces())
             .setForceArrayFields(FieldPaths.from(filterConfig.forceArrayFields()))
-            .setTypeInferenceEnabled(filterConfig.isDataTypeInferenceEnabled());
+            .setTypeInferenceEnabled(filterConfig.isDataTypeInferenceEnabled())
+            .setAttributePrefix(filterConfig.getAttributePrefix());
 
         this.reader = new XMLDocumentReader(
             filterConfig.isNamespaceAwareEnabled(),

Original file line number	Diff line number	Diff line change
`@@ -68,6 +68,8 @@ public final class XMLNodeToStructConverter implements Function<Node, TypedStruc`
`68`	`68`
`69`	`69`	`private Set<String> excludeAttributesInNamespaces = Collections.emptySet();`
`70`	`70`
	`71`	`+ private String attributePrefix = "";`
	`72`	`+`
`71`	`73`	`private FieldPaths forceArrayFields = FieldPaths.empty();`
`72`	`74`
`73`	`75`	`public XMLNodeToStructConverter setExcludeEmptyElement(boolean excludeEmptyElement) {`
`@@ -85,6 +87,11 @@ public XMLNodeToStructConverter setExcludeAttributesInNamespaces(final Set<Strin`
`85`	`87`	`return this;`
`86`	`88`	`}`
`87`	`89`
	`90`	`+ public XMLNodeToStructConverter setAttributePrefix(final String attributePrefix) {`
	`91`	`+ this.attributePrefix = attributePrefix;`
	`92`	`+ return this;`
	`93`	`+ }`
	`94`	`+`
`88`	`95`	`public XMLNodeToStructConverter setForceArrayFields(final FieldPaths forceArrayFields) {`
`89`	`96`	`this.forceArrayFields = forceArrayFields;`
`90`	`97`	`return this;`
`@@ -193,7 +200,7 @@ private Map<String, String> getNotExcludedNodeAttributes(final Node node) {`
`193`	`200`	`if (!excludeAttributesInNamespaces.contains(attr.getNamespaceURI())) {`
`194`	`201`	`String attrName = determineNodeName(attr);`
`195`	`202`	`if (isNotXmlNamespace(attr)) {`
`196`		`- attributes.put(attrName, attr.getNodeValue());`
	`203`	`+ attributes.put(attributePrefix + attrName, attr.getNodeValue());`
`197`	`204`	`}`
`198`	`205`	`}`
`199`	`206`	`}`