@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 shouldBeThreadSafeAndWorkWithManyDifferentFormatters() {
    IntStream.range(0, 10_000)
        .parallel()
        .forEach(idx -> {
          try {
            final String sourceDate = "000105X" + idx;
            final String pattern = "HHmmss'X" + idx + "'";
            final Time result = udf.parseTime(sourceDate, pattern);
            assertThat(result.getTime(), is(65000L));
          } catch (final Exception e) {
            fail(e.getMessage());
          }
        });
  }