fix: handle leap days correctly during timestamp extraction

fixes #4864, in a localized way, added a unit test case

ksqldb-common/src/main/java/io/confluent/ksql/util/timestamp/StringToTimestampParser.java

@@ -32,6 +32,7 @@ public class StringToTimestampParser {
 
   private static final Function<ZoneId, ZonedDateTime> DEFAULT_ZONED_DATE_TIME =
       zid -> ZonedDateTime.of(1970, 1, 1, 0, 0, 0, 0, zid);
+  private static final long LEAP_DAY_OF_THE_YEAR = 366;
 
   private final DateTimeFormatter formatter;
 
@@ -72,7 +73,16 @@ ZonedDateTime parseZoned(final String text, final ZoneId zoneId) {
           throw new KsqlException(
               "Unsupported temporal field in timestamp: " + text + " (" + override + ")");
         }
-        resolved = resolved.with(override, parsed.getLong(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);
       }
     }
 

ksqldb-common/src/test/java/io/confluent/ksql/util/timestamp/StringToTimestampParserTest.java

@@ -3,6 +3,7 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.is;
 
+import io.confluent.ksql.util.KsqlException;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
 import org.hamcrest.Description;
@@ -22,6 +23,9 @@ public class StringToTimestampParserTest {
   private static final ZonedDateTime FIFTH_OF_NOVEMBER =
       ZonedDateTime.of(1605, 11, 5, 0, 0, 0, 0, ZID);
 
+  private static final ZonedDateTime NEW_YEARS_EVE_2012 =
+      ZonedDateTime.of(2012, 12, 31, 23, 59, 58, 660000000, ZID);
+
   @Test
   public void shouldParseBasicLocalDate() {
     // Given
@@ -147,6 +151,29 @@ public void shouldParseDateTimeWithDayOfYear() {
     assertThat(ts, is(sameInstant(FIFTH_OF_NOVEMBER.withHour(10))));
   }
 
+  @Test
+  public void shouldParseLeapDay() {
+    // Given
+    final String format = "yyyy-MM-dd'T'HH:mm:ss.SSS";
+    final String timestamp = "2012-12-31T23:59:58.660";
+
+    // When
+    final ZonedDateTime ts = new StringToTimestampParser(format).parseZoned(timestamp, ZID);
+
+    // Then
+    assertThat(ts, is(sameInstant(NEW_YEARS_EVE_2012)));
+  }
+
+  @Test(expected = KsqlException.class)
+  public void shouldThrowErrorForLeapDayWithoutYear() {
+    // Given
+    final String format = "DDD";
+    final String timestamp = "366";
+
+    // When
+    new StringToTimestampParser(format).parseZoned(timestamp, ZID);
+  }
+
   @Test
   public void shouldResolveDefaultsForEmpty() {
     // Given