@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 shouldEncodeHexToBase64() {
    assertThat(udf.encode("4578616d706c6521", "hex", "base64"), is("RXhhbXBsZSE="));
    assertThat(udf.encode("506c616e74207472656573", "hex", "base64"), is("UGxhbnQgdHJlZXM="));
    assertThat(udf.encode("31202b2031203d2031", "hex", "base64"), is("MSArIDEgPSAx"));
    assertThat(udf.encode("ce95cebbcebbceacceb4ceb1", "hex", "base64"), is("zpXOu867zqzOtM6x"));
    assertThat(udf.encode("c39c6265726d656e736368", "hex", "base64"), is("w5xiZXJtZW5zY2g="));

    assertThat(udf.encode("0x4578616d706c6521", "hex", "base64"), is("RXhhbXBsZSE="));
    assertThat(udf.encode("X'7e8a016abfff'", "hex", "base64"), is("fooBar//"));
    assertThat(udf.encode("x'328ba7b5a8a75627b0'", "hex", "base64"), is("MountainView"));
    assertThat(udf.encode("0x", "hex", "base64"), is(""));
    assertThat(udf.encode("X''", "hex", "base64"), is(""));
    assertThat(udf.encode("x''", "hex", "base64"), is(""));
  }