@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();

    final Encode.Encoder encoder = ENCODER_MAP.get(encodedString);
    if (encoder == null) {
      throw new KsqlFunctionException("Supported input and output encodings are: "
                                  + "hex, utf8, ascii and base64");
    }
    return encoder.apply(str);
  }
@Test
  public void shouldReturnNullOnNullValue() {
    assertThat(udf.encode(null, "hex", "ascii"), is(nullValue()));
    assertThat(udf.encode(null, "utf8", "base64"), is(nullValue()));
    assertThat(udf.encode("some string", null, "utf8"), is(nullValue()));
    assertThat(udf.encode("some string", "hex", null), is(nullValue()));
  }