@Udf
  public Integer extractPort(
      @UdfParameter(description = "a valid URL to extract a port from") final String input) {
    final Integer port = UrlParser.extract(input, URI::getPort);

    // check for LT 0 because URI::getPort returns -1 if the port
    // does not exist, but UrlParser#extract will return null if
    // the URI is invalid
    return (port == null || port < 0) ? null : port;
  }
@Test
  public void shouldThrowExceptionForMalformedURL() {
    // When:
    final KsqlException e = assertThrows(
        KsqlException.class,
        () -> extractUdf.extractPort("http://257.1/bogus/[url")
    );

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