@VisibleForTesting
  ZonedDateTime parseZoned(final String text, final ZoneId zoneId) {
    final TemporalAccessor parsed = formatter.parse(text);
    final ZoneId parsedZone = parsed.query(TemporalQueries.zone());

    ZonedDateTime resolved = DEFAULT_ZONED_DATE_TIME.apply(
        ObjectUtils.defaultIfNull(parsedZone, zoneId));

    for (final TemporalField override : ChronoField.values()) {
      if (parsed.isSupported(override)) {
        if (!resolved.isSupported(override)) {
          throw new KsqlException(
              "Unsupported temporal field in timestamp: " + text + " (" + override + ")");
        }

        final long value = parsed.getLong(override);
        if (override == ChronoField.DAY_OF_YEAR && value == LEAP_DAY_OF_THE_YEAR) {
          if (!parsed.isSupported(ChronoField.YEAR)) {
            throw new KsqlException("Leap day cannot be parsed without supplying the year field");
          }
          // eagerly override year, to avoid mismatch with epoch year, which is not a leap year
          resolved = resolved.withYear(parsed.get(ChronoField.YEAR));
        }
        resolved = resolved.with(override, value);
      }
    }

    return resolved;
  }
@Test
  public void shouldParseDateTimeWithDayOfYear() {
    // Given
    final String format = "yyyy-DDD HH";
    final String timestamp = String.format("1605-%d 10", FIFTH_OF_NOVEMBER.getDayOfYear());

    // When
    final ZonedDateTime ts = new StringToTimestampParser(format).parseZoned(timestamp, ZID);

    // Then
    assertThat(ts, is(sameInstant(FIFTH_OF_NOVEMBER.withHour(10))));
  }