Merge branch '7.5.x'

Author: rayokota

client-encryption/src/test/java/io/confluent/kafka/schemaregistry/encryption/FieldEncryptionExecutorTest.java

@@ -490,8 +490,8 @@ public void testKafkaAvroSerializerRuleWithSameTag() throws Exception {
     avroSchema = avroSchema.copy(metadata, ruleSet);
     schemaRegistry.register(topic + "-value", avroSchema);
 
-    int expectedEncryptionsRule1 = 0;
-    int expectedEncryptionsRule2 = 1;
+    int expectedEncryptionsRule1 = 1;
+    int expectedEncryptionsRule2 = 0;
     RecordHeaders headers = new RecordHeaders();
     Cryptor cryptor = addSpyToCryptor(avroSerializer, "rule1");
     Cryptor cryptor2 = addSpyToCryptor(avroSerializer, "rule2");

client/src/main/java/io/confluent/kafka/schemaregistry/rules/FieldRuleExecutor.java

@@ -35,24 +35,24 @@ default Object transform(RuleContext ctx, Object message) throws RuleException {
     switch (ctx.ruleMode()) {
       case WRITE:
       case UPGRADE:
-        for (int i = ctx.index() + 1; i < ctx.rules().size(); i++) {
+        for (int i = 0; i < ctx.index(); i++) {
           Rule otherRule = ctx.rules().get(i);
           if (areTransformsWithSameTags(ctx.rule(), otherRule)) {
-            // ignore this transform if a later one has the same tags
-            log.debug("Ignoring rule '{}' during {} as rule '{}' has the same tag(s) and "
-                + "overrides it", ctx.rule().getName(), ctx.ruleMode(), otherRule.getName());
+            // ignore this transform if an earlier one has the same tags
+            log.debug("Ignoring rule '{}' during {} as rule '{}' has the same tag(s)",
+                ctx.rule().getName(), ctx.ruleMode(), otherRule.getName());
             return message;
           }
         }
         break;
       case READ:
       case DOWNGRADE:
-        for (int i = 0; i < ctx.index(); i++) {
+        for (int i = ctx.index() + 1; i < ctx.rules().size(); i++) {
           Rule otherRule = ctx.rules().get(i);
           if (areTransformsWithSameTags(ctx.rule(), otherRule)) {
-            // ignore this transform if an earlier one has the same tags
-            log.debug("Ignoring rule '{}' during {} as rule '{}' has the same tag(s) and "
-                + "overrides it", ctx.rule().getName(), ctx.ruleMode(), otherRule.getName());
+            // ignore this transform if a later one has the same tags
+            log.debug("Ignoring rule '{}' during {} as rule '{}' has the same tag(s)",
+                ctx.rule().getName(), ctx.ruleMode(), otherRule.getName());
             return message;
           }
         }

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

@@ -842,10 +842,10 @@ public void testKafkaAvroSerializerReflectionFieldTransformWithSameTag() throws
     Schema schema = createWidgetSchema();
     AvroSchema avroSchema = new AvroSchema(schema);
     Rule rule = new Rule("myRule", null, RuleKind.TRANSFORM, RuleMode.WRITE,
-        CelFieldExecutor.TYPE, ImmutableSortedSet.of("PII"), null, "value + \"-suffix\"",
+        CelFieldExecutor.TYPE, ImmutableSortedSet.of("PII"), null, "value + \"-suffix2\"",
         null, null, false);
     Rule rule2 = new Rule("myRule2", null, RuleKind.TRANSFORM, RuleMode.WRITE,
-        CelFieldExecutor.TYPE, ImmutableSortedSet.of("PII"), null, "value + \"-suffix2\"",
+        CelFieldExecutor.TYPE, ImmutableSortedSet.of("PII"), null, "value + \"-suffix\"",
         null, null, false);
     RuleSet ruleSet = new RuleSet(Collections.emptyList(), ImmutableList.of(rule, rule2));
     avroSchema = avroSchema.copy(null, ruleSet);