NIFI-9714 Added overloaded toMap to MapRecord that can convert sub-records into maps.

Signed-off-by: Matthew Burgess <mattyb149@apache.org>

This closes #5784

nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/MapRecord.java

@@ -341,7 +341,50 @@ public class MapRecord implements Record {
 
     @Override
     public Map<String, Object> toMap() {
-        return Collections.unmodifiableMap(values);
+        return toMap(false);
+    }
+
+    public Map<String, Object> toMap(boolean convertSubRecords) {
+        if (convertSubRecords) {
+            Map<String, Object> newMap = new HashMap<>();
+            values.forEach((key, value) -> {
+                Object valueToAdd;
+
+                if (value instanceof MapRecord) {
+                    valueToAdd = ((MapRecord) value).toMap(true);
+                } else if (value != null
+                        && value.getClass().isArray()
+                        && ((Object[]) value)[0] instanceof MapRecord) {
+                    Object[] records = (Object[]) value;
+                    Map<String, Object>[] maps = new Map[records.length];
+                    for (int index = 0; index < records.length; index++) {
+                        maps[index] = ((MapRecord) records[index]).toMap(true);
+                    }
+                    valueToAdd = maps;
+                } else if (value instanceof List) {
+                    List valueList = (List) value;
+                    if (!valueList.isEmpty() && valueList.get(0) instanceof MapRecord) {
+                        List<Map<String, Object>> newRecords = new ArrayList<>();
+                        for (Object o : valueList) {
+                            MapRecord rec = (MapRecord) o;
+                            newRecords.add(rec.toMap(true));
+                        }
+
+                        valueToAdd = newRecords;
+                    } else {
+                        valueToAdd = value;
+                    }
+                } else {
+                    valueToAdd = value;
+                }
+
+                newMap.put(key, valueToAdd);
+            });
+
+            return newMap;
+        } else {
+            return Collections.unmodifiableMap(values);
+        }
     }
 
     @Override

nifi-commons/nifi-record/src/test/java/org/apache/nifi/serialization/record/TestMapRecord.java

@@ -18,6 +18,8 @@
 package org.apache.nifi.serialization.record;
 
 import org.apache.nifi.serialization.SimpleRecordSchema;
+import org.apache.nifi.serialization.record.type.ArrayDataType;
+import org.apache.nifi.serialization.record.type.RecordDataType;
 import org.junit.jupiter.api.Test;
 
 import java.util.ArrayList;
@@ -30,6 +32,7 @@ import java.util.Set;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class TestMapRecord {
 
@@ -180,4 +183,54 @@ public class TestMapRecord {
         assertEquals("hello", record.getValue("bar"));
         assertEquals("hello", record.getValue("baz"));
     }
+
+    @Test
+    public void testNestedSchema() {
+        final String FOO_TEST_VAL = "test!";
+        final String NESTED_RECORD_VALUE = "Hello, world!";
+
+        final List<RecordField> fields = new ArrayList<>();
+        fields.add(new RecordField("foo", RecordFieldType.STRING.getDataType(), null, set("bar", "baz")));
+        List<RecordField> nestedFields = new ArrayList<>();
+        nestedFields.add(new RecordField("test", RecordFieldType.STRING.getDataType()));
+        RecordSchema nestedSchema = new SimpleRecordSchema(nestedFields);
+        RecordDataType nestedType = new RecordDataType(nestedSchema);
+        fields.add(new RecordField("nested", nestedType));
+        fields.add(new RecordField("list", new ArrayDataType(nestedType)));
+        RecordSchema fullSchema = new SimpleRecordSchema(fields);
+
+        Map<String, Object> nestedValues = new HashMap<>();
+        nestedValues.put("test", NESTED_RECORD_VALUE);
+        Record nestedRecord = new MapRecord(nestedSchema, nestedValues);
+        Map<String, Object> values = new HashMap<>();
+        values.put("foo", FOO_TEST_VAL);
+        values.put("nested", nestedRecord);
+
+        List<Record> list = new ArrayList<>();
+        for (int x = 0; x < 5; x++) {
+            list.add(new MapRecord(nestedSchema, nestedValues));
+        }
+        values.put("list", list);
+
+        Record record = new MapRecord(fullSchema, values);
+
+        Map<String, Object> fullConversion = ((MapRecord)record).toMap(true);
+        assertEquals(FOO_TEST_VAL, fullConversion.get("foo"));
+        assertTrue(fullConversion.get("nested") instanceof Map);
+
+        Map<String, Object> nested = (Map<String, Object>)fullConversion.get("nested");
+        assertEquals(1, nested.size());
+        assertEquals(NESTED_RECORD_VALUE, nested.get("test"));
+
+        assertTrue(fullConversion.get("list") instanceof List);
+        List recordList = (List) fullConversion.get("list");
+        assertEquals(5, recordList.size());
+        for (Object rec : recordList) {
+            assertTrue(rec instanceof Map);
+            Map<String, Object> map = (Map<String, Object>)rec;
+            assertEquals(1, map.size());
+            assertEquals(NESTED_RECORD_VALUE, map.get("test"));
+        }
+    }
+
 }