Handle records nested in arrays/maps when searching for tags (#231)

Author: rayokota

schemaregistry/serde/avro.ts

@@ -430,30 +430,38 @@ function getInlineTagsRecursively(ns: string, name: string, schema: any, tags: M
     }
   } else if (typeof schema === 'object') {
     const type = schema['type']
-    if (type === 'record') {
-      let recordNs = schema['namespace']
-      let recordName = schema['name']
-      if (recordNs === undefined) {
-        recordNs = impliedNamespace(name)
-      }
-      if (recordNs == null) {
-        recordNs = ns
-      }
-      if (recordNs !== '' && !recordName.startsWith(recordNs)) {
-        recordName = recordNs + '.' + recordName
-      }
-      const fields = schema['fields']
-      for (const field of fields) {
-        const fieldTags = field['confluent:tags']
-        const fieldName = field['name']
-        if (fieldTags !== undefined && fieldName !== undefined) {
-          tags.set(recordName + '.' + fieldName, new Set(fieldTags))
+    switch (type) {
+      case 'array':
+        getInlineTagsRecursively(ns, name, schema['items'], tags)
+        break;
+      case 'map':
+        getInlineTagsRecursively(ns, name, schema['values'], tags)
+        break;
+      case 'record':
+        let recordNs = schema['namespace']
+        let recordName = schema['name']
+        if (recordNs === undefined) {
+          recordNs = impliedNamespace(name)
         }
-        const fieldType = field['type']
-        if (fieldType !== undefined) {
-          getInlineTagsRecursively(recordNs, recordName, fieldType, tags)
+        if (recordNs == null) {
+          recordNs = ns
         }
-      }
+        if (recordNs !== '' && !recordName.startsWith(recordNs)) {
+          recordName = recordNs + '.' + recordName
+        }
+        const fields = schema['fields']
+        for (const field of fields) {
+          const fieldTags = field['confluent:tags']
+          const fieldName = field['name']
+          if (fieldTags !== undefined && fieldName !== undefined) {
+            tags.set(recordName + '.' + fieldName, new Set(fieldTags))
+          }
+          const fieldType = field['type']
+          if (fieldType !== undefined) {
+            getInlineTagsRecursively(recordNs, recordName, fieldType, tags)
+          }
+        }
+        break;
     }
   }
 }

schemaregistry/test/serde/avro.spec.ts

@@ -270,6 +270,44 @@ const unionFieldSchema = `
   "version": "1"
 }`;
 
+const complexNestedSchema = `
+{
+  "type": "record",
+    "name": "UnionTest",
+    "namespace": "test",
+    "fields": [
+    {
+      "name": "emails",
+      "type": [
+        "null",
+        {
+          "type": "array",
+          "items": {
+            "type": "record",
+            "name": "Email",
+            "fields": [
+              {
+                "name": "email",
+                "type": [
+                  "null",
+                  "string"
+                ],
+                "doc": "Email address",
+                "default": null,
+                "confluent:tags": [
+                  "PII"
+                ]
+              }
+            ]
+          }
+        }
+      ],
+      "doc": "Communication Email",
+      "default": null
+    }
+  ]
+}`;
+
 class FakeClock extends Clock {
   fixedNow: number = 0
 
@@ -1358,6 +1396,63 @@ describe('AvroSerializer', () => {
     expect(obj2.mapField).toEqual({ 'key': 'world' });
     expect(obj2.unionField).toEqual(null);
   })
+  it('complex nested encryption', async () => {
+    let conf: ClientConfig = {
+      baseURLs: [baseURL],
+      cacheCapacity: 1000
+    }
+    let client = SchemaRegistryClient.newClient(conf)
+    let serConfig: AvroSerializerConfig = {
+      useLatestVersion: true,
+      ruleConfig: {
+        secret: 'mysecret'
+      }
+    }
+    let ser = new AvroSerializer(client, SerdeType.VALUE, serConfig)
+    let dekClient = fieldEncryptionExecutor.client!
+
+    let encRule: Rule = {
+      name: 'test-encrypt',
+      kind: 'TRANSFORM',
+      mode: RuleMode.WRITEREAD,
+      type: 'ENCRYPT',
+      tags: ['PII'],
+      params: {
+        'encrypt.kek.name': 'kek1',
+        'encrypt.kms.type': 'local-kms',
+        'encrypt.kms.key.id': 'mykey',
+      },
+      onFailure: 'ERROR,NONE'
+    }
+    let ruleSet: RuleSet = {
+      domainRules: [encRule]
+    }
+
+    let info: SchemaInfo = {
+      schemaType: 'AVRO',
+      schema: complexNestedSchema,
+      ruleSet
+    }
+
+    await client.register(subject, info, false)
+
+    let obj = {
+      emails: [ {
+        email: "john@acme.com",
+      } ],
+    }
+    let bytes = await ser.serialize(topic, obj)
+
+    let deserConfig: AvroDeserializerConfig = {
+      ruleConfig: {
+        secret: 'mysecret'
+      }
+    }
+    let deser = new AvroDeserializer(client, SerdeType.VALUE, deserConfig)
+    fieldEncryptionExecutor.client = dekClient
+    let obj2 = await deser.deserialize(topic, bytes)
+    expect(obj2.emails[0].email).toEqual('john@acme.com');
+  })
   it('jsonata fully compatible', async () => {
     let rule1To2 = "$merge([$sift($, function($v, $k) {$k != 'size'}), {'height': $.'size'}])"
     let rule2To1 = "$merge([$sift($, function($v, $k) {$k != 'height'}), {'size': $.'height'}])"