code: public AuditSink create(Config config, ValueAuditRuntimeMetadata auditRuntimeMetadata) {
  
      String sinkClassName = DEFAULT_AUDIT_SINK_CLASS;
      if (config.hasPath(AUDIT_SINK_CLASS_NAME_KEY)) {
        sinkClassName = config.getString(AUDIT_SINK_CLASS_NAME_KEY);
      }
      log.info("Using audit sink class name/alias " + sinkClassName);
  
      try {
        return (AuditSink)ConstructorUtils.invokeConstructor(Class.forName(this.aliasResolver.resolve(
            sinkClassName)), config, auditRuntimeMetadata);
      } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | InstantiationException
          | ClassNotFoundException e) {
        throw new RuntimeException(e);
      }
    }
docs: Create a new {@link AuditSink} using the alias or cannonical classname specified at {@value #AUDIT_SINK_CLASS_NAME_KEY} in the <code>config</code>
  The {@link AuditSink} class MUST have an accessible constructor <code>abc(Config config, TableMetadata tableMetadata)</code>
  <br>
  If {@value #AUDIT_SINK_CLASS_NAME_KEY} is not set in <code>config</code>, a default {@link #DEFAULT_AUDIT_SINK_CLASS} is used
  
  @param config job configs
  @param auditRuntimeMetadata runtime table metadata
  
  @return a new instance of {@link AuditSink}