@Udf
  public String extractParam(
      @UdfParameter(description = "a valid URL") final String input,
      @UdfParameter(description = "the parameter key") final String paramToFind) {
    final String query = UrlParser.extract(input, URI::getQuery);
    if (query == null) {
      return null;
    }

    for (final String param : PARAM_SPLITTER.split(query)) {
      final List<String> kvParam = KV_SPLITTER.splitToList(param);
      if (kvParam.size() == 1 && kvParam.get(0).equals(paramToFind)) {
        return "";
      } else if (kvParam.size() == 2 && kvParam.get(0).equals(paramToFind)) {
        return kvParam.get(1);
      }
    }
    return null;
  }
@Test
  public void shouldExtractParamValueIfPresent() {
    assertThat(extractUdf.extractParam("https://docs.confluent.io?foo%20bar=baz%20zab&blank#scalar-functions", "foo bar"), equalTo("baz zab"));
  }