@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 shouldEncodeHexToAscii() {
    assertThat(udf.encode("4578616d706C6521", "hex", "ascii"), is("Example!"));
    assertThat(udf.encode("506C616E74207472656573", "hex", "ascii"), is("Plant trees"));
    assertThat(udf.encode("31202b2031203d2031", "hex", "ascii"), is("1 + 1 = 1"));
    assertThat(udf.encode("ce95cebbcebbceacceb4ceb1", "hex", "ascii"), is("������������"));
    assertThat(udf.encode("c39c6265726d656e736368", "hex", "ascii"), is("��bermensch"));

    assertThat(udf.encode("0x48656c6c6f20576f726c6421", "hex", "ascii"), is("Hello World!"));
    assertThat(udf.encode("0x9", "hex", "ascii"), is("\t"));
    assertThat(udf.encode("0x", "hex", "ascii"), is(""));
    assertThat(udf.encode("X'436c6f7564792a7e2a3f'", "hex", "ascii"), is("Cloudy*~*?"));
    assertThat(udf.encode("x'4578616d706C6521'", "hex", "ascii"), is("Example!"));

    assertThat(udf.encode("X''", "hex", "ascii"), is(""));
    assertThat(udf.encode("x''", "hex", "ascii"), is(""));
    assertThat(udf.encode("0x578616d706C6521", "hex", "ascii"), is("\u0005xample!"));
    Assert.assertThrows(KsqlFunctionException.class, () -> udf.encode("578616d706C6521", "hex", "ascii"));
    Assert.assertThrows(KsqlFunctionException.class, () -> udf.encode("X'578616d706C6521'", "hex", "ascii"));
    Assert.assertThrows(KsqlFunctionException.class, () -> udf.encode("x'578616d706C6521'", "hex", "ascii"));

  }