DGS-22077 Ensure tags are used during schema evolution

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

@@ -532,10 +532,7 @@
   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) {
@@ -546,7 +543,7 @@
     }
   }
 
   private Object toTransformedMessage(
       RuleContext ctx, Schema schema, Object message, FieldTransform transform) {
     FieldContext fieldCtx = ctx.currentField();
     if (schema == null) {
@@ -584,16 +581,27 @@
         if (message == null) {
           return message;
         }
-        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

@@ -425,13 +425,13 @@
   public void testKafkaAvroSerializer() throws Exception {
     IndexedRecord avroRecord = createUserRecord();
     AvroSchema avroSchema = new AvroSchema(avroRecord.getSchema());
     Rule rule = new Rule("myRule", null, RuleKind.CONDITION, RuleMode.WRITEREAD,
         CelExecutor.TYPE, null, null,
         "message.name == \"testUser\" && size(message.name) == 8 && message.kind == \"ONE\"",
         null, null, false);
     RuleSet ruleSet = new RuleSet(Collections.emptyList(), Collections.singletonList(rule));
     avroSchema = avroSchema.copy(null, ruleSet);
     schemaRegistry.register(topic + "-value", avroSchema);
 
     byte[] bytes = avroSerializer.serialize(topic, avroRecord);
     assertEquals(avroRecord, avroDeserializer.deserialize(topic, bytes));
@@ -791,7 +791,7 @@
     byte[] bytes;
     Object obj;
 
     SpecificWidget widget = new SpecificWidget("alice", 5, 1);
     Schema schema = widget.getSchema();
     AvroSchema avroSchema = new AvroSchema(schema);
     Rule rule = new Rule("myRule", null, RuleKind.TRANSFORM, RuleMode.WRITEREAD,
@@ -804,12 +804,57 @@
     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 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;