Merge branch '7.4.x' into 7.5.x by rayokota

confluentinc · Dec 9, 2023 · e713025 · e713025
2 parents 5f6a445 + 41667c8
commit e713025
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 2 deletions.
diff --git a/schema-rules/src/main/java/io/confluent/kafka/schemaregistry/rules/cel/CelExecutor.java b/schema-rules/src/main/java/io/confluent/kafka/schemaregistry/rules/cel/CelExecutor.java
@@ -162,8 +162,13 @@ protected Object execute(
       if (index >= 0) {
         String guard = expr.substring(0, index);
         if (!guard.trim().isEmpty()) {
-          Object guardResult = execute(guard, obj, args);
-          if (!Boolean.TRUE.equals(guardResult)) {
+          Object guardResult = Boolean.FALSE;
+          try {
+            guardResult = execute(guard, obj, args);
+          } catch (RuleException e) {
+            // ignore
+          }
+          if (Boolean.FALSE.equals(guardResult)) {
             // Skip the expr
             return ctx.rule().getKind() == RuleKind.CONDITION ? Boolean.TRUE : obj;
           }

diff --git a/schema-rules/src/test/java/io/confluent/kafka/schemaregistry/rules/cel/CelExecutorTest.java b/schema-rules/src/test/java/io/confluent/kafka/schemaregistry/rules/cel/CelExecutorTest.java
@@ -496,6 +496,21 @@ public void testKafkaAvroSerializerConstraintException() throws Exception {
     avroDeserializer.deserialize(topic, bytes);
   }
 
+  @Test
+  public void testKafkaAvroSerializerConstraintGuard() throws Exception {
+    IndexedRecord avroRecord = createUserRecord();
+    AvroSchema avroSchema = new AvroSchema(avroRecord.getSchema());
+    Rule rule = new Rule("myRule", null, RuleKind.CONDITION, RuleMode.READ,
+        CelExecutor.TYPE, null, null, "has(message.foo) ; message.name != \"testUser\" || 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));
+  }
+
   @Test
   public void testKafkaAvroSerializerConstraintIgnore() throws Exception {
     IndexedRecord avroRecord = createUserRecord();