class Encode {

  @Udf(description = "Returns a new string encoded using the outputEncoding ")
  public String encode(
      @UdfParameter(
          description = "The source string. If null, then function returns null.") final String str,
      @UdfParameter(
          description = "The input encoding."
              + " If null, then function returns null.") final String inputEncoding,
      @UdfParameter(
          description = "The output encoding."
              + " If null, then function returns null.") final String outputEncoding) {
    if (str == null || inputEncoding == null || outputEncoding == null) {
      return null;
    }

    final String encodedString = inputEncoding.toLowerCase() + outputEncoding.toLowerCase();

    if (ENCODER_MAP.get(encodedString) == null) {
      throw new KsqlFunctionException("Supported input and output encodings are: "
                                  + "hex, utf8, ascii and base64");
    }
    return ENCODER_MAP.get(encodedString).apply(str);
  }

}