@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 shouldThrowExceptionForMalformedURL() {
    // When:
    final KsqlException e = assertThrows(
        KsqlException.class,
        () -> extractUdf.extractParam("http://257.1/bogus/[url", "foo bar")
    );

    // Given:
    assertThat(e.getMessage(), containsString("URL input has invalid syntax: http://257.1/bogus/[url"));
  }