Skip to content

Commit

Permalink
feat: read attributes from root node of XML documents (#4)
Browse files Browse the repository at this point in the history 
This commit modifies the Converter so that it now captures
attributes from the root nodes of XML documents. The logic change
for the Converter is relatively minor (read the attributes from
the XML reader before starting to recursively traverse the
document) so most of the changes in this commit are to add a few
variations of test to verify the new behaviour.

Signed-off-by: Dale Lane <[email protected]>
  • Loading branch information
@dalelane
dalelane authored Sep 2, 2024
1 parent 087c056 commit 774efc4
Show file tree
Hide file tree
Showing 19 changed files with 505 additions and 30 deletions.
22 changes: 11 additions & 11 deletions pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,13 +8,13 @@
<groupId>com.ibm.eventstreams.kafkaconnect.plugins</groupId>
<artifactId>kafka-connect-xml-converter</artifactId>
<description>Kafka Connect plugins for processing XML data</description>
<version>0.1.0</version>
<version>0.2.0</version>

<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>connect-api</artifactId>
<version>3.5.1</version>
<version>3.7.1</version>
<scope>provided</scope>
</dependency>
<dependency>
Expand All @@ -25,7 +25,7 @@
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>2.0.9</version>
<version>2.0.16</version>
</dependency>

<dependency>
Expand All @@ -49,31 +49,31 @@
<dependency>
<groupId>org.xmlunit</groupId>
<artifactId>xmlunit-core</artifactId>
<version>2.9.1</version>
<version>2.10.0</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.xmlunit</groupId>
<artifactId>xmlunit-matchers</artifactId>
<version>2.9.1</version>
<version>2.10.0</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>connect-json</artifactId>
<version>3.6.0</version>
<version>3.7.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.15.3</version>
<version>2.17.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-nop</artifactId>
<version>2.0.9</version>
<version>2.0.16</version>
<scope>test</scope>
</dependency>
</dependencies>
Expand All @@ -91,7 +91,7 @@
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.11.0</version>
<version>3.13.0</version>
<inherited>true</inherited>
<configuration>
<source>17</source>
Expand All @@ -103,7 +103,7 @@
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.6.0</version>
<version>3.7.1</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
Expand All @@ -122,7 +122,7 @@
<plugin>
<groupId>org.jacoco</groupId>
<artifactId>jacoco-maven-plugin</artifactId>
<version>0.8.10</version>
<version>0.8.12</version>
<configuration>
<excludes>
<!-- requires a queue manager to test, so excluded from unit test coverage reports -->
Expand Down
22 changes: 14 additions & 8 deletions src/main/java/com/ibm/eventstreams/kafkaconnect/plugins/xml/engines/XStreamMapConverter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -45,16 +45,26 @@ public boolean canConvert(Class type) {

@Override
public Object unmarshal(HierarchicalStreamReader reader, UnmarshallingContext context) {
return xmlToMap(reader);
return xmlToMap(reader, true);
}



private Map<String, Object> xmlToMap(HierarchicalStreamReader reader, Map<String, Object> map) {
private Map<String, Object> xmlToMap(HierarchicalStreamReader reader, boolean isXmlRoot) {
final List<Map<String, Object>> list = new ArrayList<Map<String, Object>>();

Map<String, Object> attrs = new LinkedHashMap<>();

if (isXmlRoot) {
// capture attributes from the root node of the XML document
// (not needed when recursing deeper in the document as these
// will have already been captured)
attrs = attrsToMap(reader);
if (attrs.size() > 0) {
list.add(attrs);
}
}

while (reader.hasMoreChildren()) {
// get attributes of current node
attrs = attrsToMap(reader);
Expand All @@ -64,7 +74,7 @@ private Map<String, Object> xmlToMap(HierarchicalStreamReader reader, Map<String

if (reader.hasMoreChildren()) {
// child nodes found at the lower level - process recursively
final Map<String, Object> innerItem = xmlToMap(reader);
final Map<String, Object> innerItem = xmlToMap(reader, false);

if (attrs.size() > 0) {
mergeAttributesIntoNode(reader.getNodeName(), innerItem, attrs);
Expand Down Expand Up @@ -102,6 +112,7 @@ private Map<String, Object> xmlToMap(HierarchicalStreamReader reader, Map<String
}

// return as a map
Map<String, Object> map = new LinkedHashMap<>();
map.put(reader.getNodeName(), convertMapToList(flatten(list)));

return map;
Expand Down Expand Up @@ -135,11 +146,6 @@ private Map<String, Object> attrsToMap(HierarchicalStreamReader reader) {
}


private Map<String, Object> xmlToMap(HierarchicalStreamReader reader) {
return xmlToMap(reader, new LinkedHashMap<String, Object>());
}


private boolean isEntry(Map<String, ?> map) {
return map.size() == 1 && map.containsKey("entry");
}
Expand Down
18 changes: 14 additions & 4 deletions ...in/java/com/ibm/eventstreams/kafkaconnect/plugins/xml/engines/XStreamStructConverter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ public void marshal(Object source, HierarchicalStreamWriter writer, MarshallingC
public Object unmarshal(HierarchicalStreamReader reader, UnmarshallingContext context) {
if (schema.type() == Type.STRUCT) {
final Struct newObject = new Struct(schema);
xmlToStruct(reader, schema, newObject);
xmlToStruct(reader, schema, newObject, true);
return newObject;
}
else if (schema.type() == Type.MAP) {
Expand Down Expand Up @@ -280,7 +280,17 @@ else if (valueSchema.type() == Type.MAP) {



private void xmlToStruct(HierarchicalStreamReader reader, Schema schema, Struct object) {
private void xmlToStruct(HierarchicalStreamReader reader, Schema schema, Struct object, boolean isXmlRoot) {

if (isXmlRoot) {
// capture attributes from the root node of the XML document
// (not needed when recursing deeper in the document as these
// will have already been captured)
Map<String, String> rootAttrs = getAttributes(reader);
if (rootAttrs.size() > 0) {
processStruct(reader, object, rootAttrs, schema);
}
}

while (reader.hasMoreChildren()) {

Expand Down Expand Up @@ -349,7 +359,7 @@ private void processStruct(HierarchicalStreamReader reader, Struct struct, Map<S
});

// populate the struct
xmlToStruct(reader, schema, struct);
xmlToStruct(reader, schema, struct, false);

// add any attributes from the node to the struct
addAttributesToStruct(attributes, struct, schema);
Expand Down Expand Up @@ -394,7 +404,7 @@ private void put(List<Object> list, Schema listItemSchema, String valueStr, Map<
}

if (reader.hasMoreChildren()) {
xmlToStruct(reader, listItemSchema, obj);
xmlToStruct(reader, listItemSchema, obj, false);
}

list.add(obj);
Expand Down
5 changes: 4 additions & 1 deletion src/test/java/com/ibm/eventstreams/kafkaconnect/plugins/xml/XmlConverterSinkTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,10 @@ public static Collection<Object[]> testCases() {
{ "038", true, true, true },
// { "039", true, true, true },
{ "040", true, true, true },
{ "047", false, true, false }
{ "047", false, true, false },
{ "052", false, false, false },
{ "053", false, false, false },
{ "054", false, true, false }
});
}

Expand Down
5 changes: 4 additions & 1 deletion src/test/java/com/ibm/eventstreams/kafkaconnect/plugins/xml/XmlTransformationStringTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,10 @@ public static Collection<Object[]> testCases() {
{ "025", true, true },
{ "027", false, true },
{ "028", true, true },
{ "037", false, false }
{ "037", false, false },
{ "052", true, true },
{ "053", true, true },
{ "054", false, true }
});
}

Expand Down
5 changes: 4 additions & 1 deletion ...t/java/com/ibm/eventstreams/kafkaconnect/plugins/xml/engines/XStreamMapConverterTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -79,10 +79,13 @@ public static Collection<Object[]> testCases() {
// { "034", true },
// { "035", true },
// { "036", true },
{ "037", true }
{ "037", true },
// { "038", true },
// { "039", true },
// { "040", true },
{ "052", false },
{ "053", false },
{ "054", true }
});
}

Expand Down
5 changes: 4 additions & 1 deletion ...ava/com/ibm/eventstreams/kafkaconnect/plugins/xml/engines/XStreamStructConverterTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,10 @@ public static Collection<Object[]> testCases() {
{ "040", false },
{ "042", false },
{ "043", false },
{ "044", false }
{ "044", false },
{ "052", false },
{ "053", false },
{ "054", false }
});
}

Expand Down
5 changes: 4 additions & 1 deletion src/test/java/com/ibm/eventstreams/kafkaconnect/plugins/xml/engines/XmlToStructTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,10 @@ public static Collection<Object[]> testCases() {
{ "041", true, true },
{ "042", true, true },
{ "043", true, false },
{ "047", false, false }
{ "047", false, false },
{ "052", false, false },
{ "053", false, false },
{ "054", true, false }
});
}

Expand Down
5 changes: 4 additions & 1 deletion src/test/java/com/ibm/eventstreams/kafkaconnect/plugins/xml/engines/XsdToSchemaTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,10 @@ public static Collection<Object[]> testCases() {
{ "037", false },
{ "038", true },
{ "040", true },
{ "047", false }
{ "047", false },
{ "052", false },
{ "053", false },
{ "054", false }
});
}

Expand Down
90 changes: 89 additions & 1 deletion src/test/java/com/ibm/eventstreams/kafkaconnect/plugins/xml/testutils/MapGenerators.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,7 @@ public class MapGenerators {
case "005":
value.put("entry", ListGenerators.get(testCaseId));
break;
case "006":
case "006": {
final Map<String, Object> test1 = new LinkedHashMap<>();
test1.put("test2", "one");
final Map<String, Object> test5_0 = new LinkedHashMap<>();
Expand All @@ -137,6 +137,7 @@ public class MapGenerators {
value.put("test1", test1);
value.put("outer", outer);
break;
}
case "007":
value.put("type0", "string");
value.put("type1", "https://www.ibm.com");
Expand Down Expand Up @@ -968,6 +969,65 @@ public class MapGenerators {
value.put("map-of-arrays", mapOfArrays);
break;
}
case "052": {
value.put("myattr1", "testing");
value.put("myattr2", 123);
value.put("test1", "one");
value.put("test2", "two");
break;
}
case "053": {
value.put("myattr1", "testing");
value.put("myattr2", 123);
value.put("myattr3", false);

Map<String, Object> test1 = new LinkedHashMap<>();
test1.put("myattr4", "inner");
test1.put("entry", "one");
value.put("test1", test1);

Map<String, Object> test2 = new LinkedHashMap<>();
test2.put("test2a", "alpha");
Map<String, Object> test2b = new LinkedHashMap<>();
test2b.put("myattr5", "deepinner");
test2b.put("entry", "beta");
test2.put("test2b", test2b);
value.put("test2", test2);

Map<String, Object> test3 = new LinkedHashMap<>();
test3.put("myattr6", "middle");
test3.put("test3a", 333);
value.put("test3", test3);

value.put("test4", "four");
break;
}
case "054": {
Map<String, Object> transaccion = new LinkedHashMap<>();
transaccion.put("id", "PE80");
transaccion.put("tecla", "00");
transaccion.put("numclie", 51372133);
Map<String, Object> datos = new LinkedHashMap<>();
datos.put("transaccion", transaccion);
Map<String, Object> xmlentrada = new LinkedHashMap<>();
xmlentrada.put("datos", datos);
Map<String, Object> data = new LinkedHashMap<>();
data.put("xml-entrada", xmlentrada);
data.put("trama-entrada", "longer test string");
data.put("mq-server", "MQ : SPIAWT99;;;SPIA.QC.QZT1;SPIA.QP.OUT");
data.put("direccion-IP", "17.127.22.33");
data.put("nombre-servidor", "Qpbxiaa");
data.put("canal", "ABC");

value.put("date", "2024-08-06 12:32:04");
value.put("cr", 2246);
value.put("tx", "PE23");
value.put("user", "01888329");
value.put("estatus-tx", 1);
value.put("version", "7.0.1001.2");
value.put("data", data);
break;
}
}
return value;
}
Expand Down Expand Up @@ -1589,6 +1649,34 @@ public class MapGenerators {
value.put("base64EncodedBytes", "U2VjcmV0IG1lc3NhZ2U=");
return value;
}
case "054": {
final Map<String, Object> value = new LinkedHashMap<>();

Map<String, Object> transaccion = new LinkedHashMap<>();
transaccion.put("id", "PE80");
transaccion.put("tecla", 0);
transaccion.put("numclie", 51372133);
Map<String, Object> datos = new LinkedHashMap<>();
datos.put("transaccion", transaccion);
Map<String, Object> xmlentrada = new LinkedHashMap<>();
xmlentrada.put("datos", datos);
Map<String, Object> data = new LinkedHashMap<>();
data.put("xml-entrada", xmlentrada);
data.put("trama-entrada", "longer test string");
data.put("mq-server", "MQ : SPIAWT99;;;SPIA.QC.QZT1;SPIA.QP.OUT");
data.put("direccion-IP", "17.127.22.33");
data.put("nombre-servidor", "Qpbxiaa");
data.put("canal", "ABC");

value.put("date", "2024-08-06 12:32:04");
value.put("cr", 2246);
value.put("tx", "PE23");
value.put("user", 1888329);
value.put("estatus-tx", 1);
value.put("version", "7.0.1001.2");
value.put("data", data);
return value;
}
default:
return get(testCaseId);
}
Expand Down
Loading

0 comments on commit 774efc4

Please sign in to comment.