Add ability to override disable flag and actions on a rule (#200)

* Add ability to override disable flag and actions on a rule

* Add test

Author: rayokota

schemaregistry/serde/rule-registry.ts

@@ -1,11 +1,22 @@
 import {RuleAction, RuleExecutor} from "./serde";
 
+/**
+ * RuleOverride represents a rule override
+ */
+export interface RuleOverride {
+  type: string
+  onSuccess?: string
+  onFailure?: string
+  disabled?: boolean
+}
+
 /**
  * RuleRegistry is used to register and fetch rule executors and actions.
  */
 export class RuleRegistry {
   private ruleExecutors: Map<string, RuleExecutor> = new Map<string, RuleExecutor>()
   private ruleActions: Map<string, RuleAction> = new Map<string, RuleAction>()
+  private ruleOverrides: Map<string, RuleOverride> = new Map<string, RuleOverride>()
 
   private static globalInstance: RuleRegistry = new RuleRegistry()
 
@@ -55,12 +66,36 @@ export class RuleRegistry {
     return Array.from(this.ruleActions.values())
   }
 
+  /**
+   * registerOverride is used to register a new rule override.
+   * @param ruleOverride - the rule override to register
+   */
+  public registerOverride(ruleOverride: RuleOverride): void {
+    this.ruleOverrides.set(ruleOverride.type, ruleOverride)
+  }
+
+  /**
+   * getOverride fetches a rule override by a given name.
+   * @param name - the name of the rule override to fetch
+   */
+  public getOverride(name: string): RuleOverride | undefined {
+    return this.ruleOverrides.get(name)
+  }
+
+  /**
+   * getOverrides fetches all rule overrides
+   */
+  public getOverrides(): RuleOverride[] {
+    return Array.from(this.ruleOverrides.values())
+  }
+
   /**
    * clear clears all registered rules
    */
   public clear(): void {
     this.ruleExecutors.clear()
     this.ruleActions.clear()
+    this.ruleOverrides.clear()
   }
 
   /**
@@ -85,4 +120,12 @@ export class RuleRegistry {
   public static registerRuleAction(ruleAction: RuleAction): void {
     RuleRegistry.globalInstance.registerAction(ruleAction)
   }
+
+  /**
+   * registerRuleOverride is used to register a new rule override globally.
+   * @param ruleOverride - the rule override to register
+   */
+  public static registerRuleOverride(ruleOverride: RuleOverride): void {
+    RuleRegistry.globalInstance.registerOverride(ruleOverride)
+  }
 }

schemaregistry/serde/serde.ts

@@ -136,7 +136,7 @@ export abstract class Serde {
         this.serdeType === SerdeType.KEY, ruleMode, rule, i, rules, inlineTags, this.fieldTransformer!)
       let ruleExecutor = this.ruleRegistry.getExecutor(rule.type)
       if (ruleExecutor == null) {
-        await this.runAction(ctx, ruleMode, rule, rule.onFailure, msg,
+        await this.runAction(ctx, ruleMode, rule, this.getOnFailure(rule), msg,
           new Error(`could not find rule executor of type ${rule.type}`), 'ERROR')
         return msg
       }
@@ -152,18 +152,43 @@ export abstract class Serde {
             msg = result
             break
         }
-        await this.runAction(ctx, ruleMode, rule, msg != null ? rule.onSuccess : rule.onFailure,
+        await this.runAction(ctx, ruleMode, rule, msg != null
+            ? this.getOnSuccess(rule) : this.getOnFailure(rule),
           msg, null, msg != null ? 'NONE' : 'ERROR')
       } catch (error) {
         if (error instanceof SerializationError) {
           throw error
         }
-        await this.runAction(ctx, ruleMode, rule, rule.onFailure, msg, error as Error, 'ERROR')
+        await this.runAction(ctx, ruleMode, rule, this.getOnFailure(rule), msg, error as Error, 'ERROR')
       }
     }
     return msg
   }
 
+  getOnSuccess(rule: Rule): string | undefined {
+    let override = this.ruleRegistry.getOverride(rule.type)
+    if (override != null && override.onSuccess != null) {
+      return override.onSuccess
+    }
+    return rule.onSuccess
+  }
+
+  getOnFailure(rule: Rule): string | undefined {
+    let override = this.ruleRegistry.getOverride(rule.type)
+    if (override != null && override.onFailure != null) {
+      return override.onFailure
+    }
+    return rule.onFailure
+  }
+
+  isDisabled(rule: Rule): boolean | undefined {
+    let override = this.ruleRegistry.getOverride(rule.type)
+    if (override != null && override.disabled != null) {
+      return override.disabled
+    }
+    return rule.disabled
+  }
+
   async runAction(ctx: RuleContext, ruleMode: RuleMode, rule: Rule, action: string | undefined,
             msg: any, err: Error | null, defaultAction: string): Promise<void> {
     let actionName = this.getRuleActionName(rule, ruleMode, action)

schemaregistry/test/serde/avro.spec.ts

@@ -659,9 +659,17 @@ describe('AvroSerializer', () => {
     expect(obj2.boolField).toEqual(obj.boolField);
     expect(obj2.bytesField).toEqual(obj.bytesField);
 
-    clearKmsClients()
     let registry = new RuleRegistry()
     registry.registerExecutor(new FieldEncryptionExecutor())
+    registry.registerOverride({type: 'ENCRYPT', disabled: true})
+    deser = new AvroDeserializer(client, SerdeType.VALUE, deserConfig, registry)
+    obj2 = await deser.deserialize(topic, bytes)
+    expect(obj2.stringField).not.toEqual(obj.stringField);
+    expect(obj2.bytesField).not.toEqual(obj.bytesField);
+
+    clearKmsClients()
+    registry = new RuleRegistry()
+    registry.registerExecutor(new FieldEncryptionExecutor())
     deser = new AvroDeserializer(client, SerdeType.VALUE, {}, registry)
     obj2 = await deser.deserialize(topic, bytes)
     expect(obj2.stringField).not.toEqual(obj.stringField);