@Udf(description = "Converts a string representation of a time in the given format"
      + " into the TIME value.")
  public Time parseTime(
      @UdfParameter(
          description = "The string representation of a time.") final String formattedTime,
      @UdfParameter(
          description = "The format pattern should be in the format expected by"
              + " java.time.format.DateTimeFormatter.") final String formatPattern) {
    if (formattedTime == null | formatPattern == null) {
      return null;
    }
    try {
      final TemporalAccessor ta = formatters.get(formatPattern).parse(formattedTime);
      final Optional<ChronoField> dateField = Arrays.stream(ChronoField.values())
          .filter(ChronoField::isDateBased)
          .filter(ta::isSupported)
          .findFirst();

      if (dateField.isPresent()) {
        throw new KsqlFunctionException("Time format contains date field.");
      }

      return new Time(TimeUnit.NANOSECONDS.toMillis(LocalTime.from(ta).toNanoOfDay()));
    } catch (ExecutionException | RuntimeException e) {
      throw new KsqlFunctionException("Failed to parse time '" + formattedTime
          + "' with formatter '" + formatPattern
          + "': " + e.getMessage(), e);
    }
  }
@Test
  public void shouldThrowOnEmptyString() {
    // When:
    final Exception e = assertThrows(
        KsqlFunctionException.class,
        () -> udf.parseTime("", "HHmmss")
    );

    // Then:
    assertThat(e.getMessage(), containsString("Failed to parse time '' with formatter 'HHmmss'"));
  }