fix: ignore null values and not existing targer fields

Author: max-prosper

README.md

@@ -3,7 +3,7 @@ This lib implements Kafka connect SMT (Single Message Transformation) to
 convert values of specified fields to JSON strings (stringify).
 
 ## Config
-Use it in connector config file like this:
+Use it in connector config file:
 ~~~json
 ...
 "transforms": "stringify",
@@ -15,23 +15,23 @@ Use it in connector config file like this:
 Use dot notation for deeper fields (e. g. `level1.level2`).
 
 ## Install to Kafka Connect
-After build copy file `target/stirngify-json-smt-0.0.1-jar-with-deps.jar`
+After build copy file `target/stirngify-json-smt-0.0.2-jar-with-deps.jar`
 to Kafka Connect container `` copying to its docker image or so.
 
 It can be done adding this line to Dockerfile:
 ~~~Dockerfile
-COPY ./target/stringify-json-smt-0.0.1-jar-with-deps.jar $KAFKA_CONNECT_PLUGINS_DIR
+COPY ./target/stringify-json-smt-0.0.2-jar-with-deps.jar $KAFKA_CONNECT_PLUGINS_DIR
 ~~~
 
 Or download current release:
 ~~~Dockerfile
 RUN curl -fSL -o /tmp/plugin.tar.gz \
-    https://github.com/max-prosper/stringify-json-smt/releases/download/0.0.1/stringify-json-smt-0.0.1.tar.gz && \
+    https://github.com/max-prosper/stringify-json-smt/releases/download/0.0.2/stringify-json-smt-0.0.2.tar.gz && \
     tar -xzf /tmp/plugin.tar.gz -C $KAFKA_CONNECT_PLUGINS_DIR && \
     rm -f /tmp/plugin.tar.gz;
 ~~~
 
 ## Build release file
-- Increment version in `pom.xml` (e.g. to `0.0.2`).
-- Run build script: `./scripts/build-release.sh 0.0.2`.
+- Increment version in `pom.xml` (e.g. to `0.0.3`).
+- Run build script: `./scripts/build-release.sh 0.0.3`.
 - Take `*.tar.gz` file from `target` folder and publish it.

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.github.maxprosper.smt.stringifyjson</groupId>
     <artifactId>stringify-json-smt</artifactId>
-    <version>0.0.1</version>
+    <version>0.0.2</version>
 
     <properties>
         <commons-lang3.version>3.8.1</commons-lang3.version>

src/main/java/com/github/maxprosper/smt/stringifyjson/StringifyJson.java

@@ -88,11 +88,22 @@ public R apply(R record) {
     @SuppressWarnings("unchecked")
     private static HashMap<String, String> stringifyFields(Struct value, List<String> targetFields) {
         final HashMap<String, String> result = new HashMap<>(targetFields.size());
+        final Schema valueSchema = value.schema();
 
         for (String field : targetFields) {
             String[] pathArr = field.split("\\.");
             List<String> path = Arrays.asList(pathArr);
+
+            if (valueSchema.field(field) == null) {
+                LOGGER.warn("target field {} not present in the record schema", field);
+                continue;
+            }
+
             Object fieldValue = getFieldValue(path, value);
+            if (fieldValue == null) {
+                LOGGER.info("target field {} is null, nothing to stringify", field);
+                continue;
+            }
 
             String strValue;
             Schema.Type fieldValueType = Values.inferSchema(fieldValue).type();
@@ -188,8 +199,8 @@ public static String arrayValueToString(List<Object> value) {
             }
             Schema valueSchema = Values.inferSchema(elem);
             if (valueSchema == null) {
-               builder.append("null");
-               continue;
+                builder.append("null");
+                continue;
             }
             Schema.Type valueType = valueSchema.type();
             if (valueType.equals(Schema.Type.STRUCT)) {

src/test/java/com/github/maxprosper/smt/stringifyjson/StringifyJsonTest.java

@@ -54,6 +54,42 @@ public void noSchema() {
         Assertions.assertNull(output);
     }
 
+    @Test
+    public void nullTargetField() {
+        final Schema valueSchema = SchemaBuilder.struct()
+                .field("target_field", Schema.OPTIONAL_STRING_SCHEMA)
+                .field("other_field", Schema.INT32_SCHEMA);
+
+        final Map<String, String> props = new HashMap<>();
+        props.put(propTargetFields, "target_field");
+        xform.configure(props);
+
+        final Struct inputValue = new Struct(valueSchema)
+                .put("other_field", 24);
+
+        final String outputValue = "Struct{other_field=24}";
+        final String outputSchema = "[Field{name=target_field, index=0, schema=Schema{STRING}}, Field{name=other_field, index=1, schema=Schema{INT32}}]";
+
+        runAssertions(valueSchema, inputValue, outputSchema, outputValue);
+    }
+
+    @Test
+    public void targetFieldNotPresentInTheSchema() {
+        final Schema valueSchema = SchemaBuilder.struct()
+                .field("other_field", Schema.INT32_SCHEMA);
+
+        final Map<String, String> props = new HashMap<>();
+        props.put(propTargetFields, "target_field");
+        xform.configure(props);
+
+        final Struct inputValue = new Struct(valueSchema)
+                .put("other_field", 24);
+
+        final String outputValue = "Struct{other_field=24}";
+        final String outputSchema = "[Field{name=other_field, index=0, schema=Schema{INT32}}]";
+
+        runAssertions(valueSchema, inputValue, outputSchema, outputValue);
+    }
     @Test
     public void integerField() {
         final Schema valueSchema = SchemaBuilder.struct()