class ArrayContainsKudf implements Kudf {

  @Override
  public Object evaluate(Object... args) {
    if (args.length != 2) {
      throw new KsqlFunctionException("ARRAY_CONTAINS udf should have two input argument. "
          + "Given: " + Arrays.toString(args));
    }
    Object searchValue = args[1];
    if (args[0] instanceof String) {
      return jsonStringArrayContains(searchValue, (String) args[0]);
    } else if (args[0] instanceof Object[]) {
      return ArrayUtil.containsValue(searchValue, (Object[]) args[0]);
    }
    throw new KsqlFunctionException("Invalid type parameters for " + Arrays.toString(args));
  }

    ArrayContainsKudf();

}