DGS-22077 Ensure tags are used during schema evolution (#3890) (#3892)

Author: rayokota

client/src/main/java/io/confluent/kafka/schemaregistry/avro/AvroSchema.java

@@ -561,10 +561,7 @@ public Object copyMessage(Object message) {
   public Object transformMessage(RuleContext ctx, FieldTransform transform, Object message)
       throws RuleException {
     try {
-      // Use the schema from the message if it exists, so schema evolution works properly
-      Schema schema = message instanceof GenericContainer
-          ? ((GenericContainer) message).getSchema()
-          : this.rawSchema();
+      Schema schema = this.rawSchema();
       return toTransformedMessage(ctx, schema, message, transform);
     } catch (RuntimeException e) {
       if (e.getCause() instanceof RuleException) {
@@ -613,16 +610,27 @@ private Object toTransformedMessage(
         if (message == null) {
           return null;
         }
-        data = AvroSchemaUtils.getData(schema, message, false, false);
-        for (Schema.Field f : schema.getFields()) {
-          String fullName = schema.getFullName() + "." + f.name();
+        Schema recordSchema = schema;
+        if (message instanceof GenericContainer) {
+          // Use the schema from the message if it exists, so schema evolution works properly
+          recordSchema = ((GenericContainer) message).getSchema();
+        }
+        data = AvroSchemaUtils.getData(recordSchema, message, false, false);
+        for (Schema.Field f : recordSchema.getFields()) {
+          // The original field has tags needed for inline tag matching
+          Schema.Field originalField = schema.getField(f.name());
+          if (originalField == null) {
+            originalField = f;
+          }
+          String fullName = recordSchema.getFullName() + "." + f.name();
           try (FieldContext fc = ctx.enterField(
-              message, fullName, f.name(), getType(f.schema()), getInlineTags(f))) {
+              message, fullName, f.name(),
+              getType(originalField.schema()), getInlineTags(originalField))) {
             Object value = data.getField(message, f.name(), f.pos());
             if (value instanceof Utf8) {
               value = value.toString();
             }
-            Object newValue = toTransformedMessage(ctx, f.schema(), value, transform);
+            Object newValue = toTransformedMessage(ctx, originalField.schema(), value, transform);
             if (ctx.rule().getKind() == RuleKind.CONDITION) {
               if (Boolean.FALSE.equals(newValue)) {
                 throw new RuntimeException(new RuleConditionException(ctx.rule()));

schema-rules/src/test/java/io/confluent/kafka/schemaregistry/rules/cel/CelExecutorTest.java

@@ -810,6 +810,51 @@ public void testKafkaAvroSerializerSpecificFieldTransform() throws Exception {
     assertEquals("alice-suffix-suffix", ((SpecificWidget)obj).getName().toString());
   }
 
+  @Test
+  public void testKafkaAvroSerializerSpecificFieldTransformWithTag() throws Exception {
+    byte[] bytes;
+    Object obj;
+
+    String schemaStr = "{\n"
+        + "  \"namespace\": \"io.confluent.kafka.schemaregistry.rules\",\n"
+        + "  \"type\": \"record\",\n"
+        + "  \"name\": \"SpecificWidget\",\n"
+        + "  \"fields\": [\n"
+        + "    {\n"
+        + "      \"name\": \"name\",\n"
+        + "      \"type\": \"string\",\n"
+        + "      \"confluent:tags\": [ \"PII\" ]\n"
+        + "    },\n"
+        + "    {\n"
+        + "      \"name\": \"size\",\n"
+        + "      \"type\": \"int\"\n"
+        + "    },\n"
+        + "    {\n"
+        + "      \"name\": \"version\",\n"
+        + "      \"type\": \"int\"\n"
+        + "    }\n"
+        + "  ]\n"
+        + "}";
+
+    Schema schema = new Schema.Parser().parse(schemaStr);
+    AvroSchema avroSchema = new AvroSchema(schema);
+    Rule rule = new Rule("myRule", null, RuleKind.TRANSFORM, RuleMode.WRITEREAD,
+        CelFieldExecutor.TYPE, Collections.singleton("PII"), null, "value + \"-suffix\"",
+        null, null, false);
+    RuleSet ruleSet = new RuleSet(Collections.emptyList(), Collections.singletonList(rule));
+    avroSchema = avroSchema.copy(null, ruleSet);
+    schemaRegistry.register(topic + "-value", avroSchema);
+
+    SpecificWidget widget = new SpecificWidget("alice", 5, 1);
+    bytes = specificAvroSerializer.serialize(topic, widget);
+    obj = specificAvroDeserializer.deserialize(topic, bytes);
+    assertTrue(
+        "Returned object does not match",
+        SpecificWidget.class.isInstance(obj)
+    );
+    assertEquals("alice-suffix-suffix", ((SpecificWidget)obj).getName().toString());
+  }
+
   @Test
   public void testKafkaAvroSerializerSpecificFieldTransformWithMissingProp() throws Exception {
     byte[] bytes;