Allow for negative epoch seconds (elastic#111938)

The change to allow nanoseconds in ZonedDateTime split the epoch seconds
from the nanosecond subelement. However, the epoch seconds were then
written as a vlong, when in fact they could be negative if the date is
before epoch. This commit changes the format to use zlong instead, which
supports negatives.

closes elastic#111923

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -189,10 +189,11 @@ static TransportVersion def(int id) {
     public static final TransportVersion ESQL_ORIGINAL_INDICES = def(8_719_00_0);
     public static final TransportVersion ML_INFERENCE_EIS_INTEGRATION_ADDED = def(8_720_00_0);
     public static final TransportVersion INGEST_PIPELINE_EXCEPTION_ADDED = def(8_721_00_0);
-    public static final TransportVersion ZDT_NANOS_SUPPORT = def(8_722_00_0);
+    public static final TransportVersion ZDT_NANOS_SUPPORT_BROKEN = def(8_722_00_0);
     public static final TransportVersion REMOVE_GLOBAL_RETENTION_FROM_TEMPLATES = def(8_723_00_0);
     public static final TransportVersion RANDOM_RERANKER_RETRIEVER = def(8_724_00_0);
     public static final TransportVersion ESQL_PROFILE_SLEEPS = def(8_725_00_0);
+    public static final TransportVersion ZDT_NANOS_SUPPORT = def(8_726_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/common/io/stream/StreamInput.java

@@ -903,8 +903,11 @@ public final Instant readOptionalInstant() throws IOException {
     private ZonedDateTime readZonedDateTime() throws IOException {
         final String timeZoneId = readString();
         final Instant instant;
-        if (getTransportVersion().onOrAfter(TransportVersions.ZDT_NANOS_SUPPORT)) {
-            instant = Instant.ofEpochSecond(readVLong(), readInt());
+        if (getTransportVersion().onOrAfter(TransportVersions.ZDT_NANOS_SUPPORT_BROKEN)) {
+            // epoch seconds can be negative, but it was incorrectly first written as vlong
+            boolean zlong = getTransportVersion().onOrAfter(TransportVersions.ZDT_NANOS_SUPPORT);
+            long seconds = zlong ? readZLong() : readVLong();
+            instant = Instant.ofEpochSecond(seconds, readInt());
         } else {
             instant = Instant.ofEpochMilli(readLong());
         }

server/src/main/java/org/elasticsearch/common/io/stream/StreamOutput.java

@@ -767,8 +767,13 @@ public final void writeOptionalInstant(@Nullable Instant instant) throws IOExcep
             final ZonedDateTime zonedDateTime = (ZonedDateTime) v;
             o.writeString(zonedDateTime.getZone().getId());
             Instant instant = zonedDateTime.toInstant();
-            if (o.getTransportVersion().onOrAfter(TransportVersions.ZDT_NANOS_SUPPORT)) {
-                o.writeVLong(instant.getEpochSecond());
+            if (o.getTransportVersion().onOrAfter(TransportVersions.ZDT_NANOS_SUPPORT_BROKEN)) {
+                // epoch seconds can be negative, but it was incorrectly first written as vlong
+                if (o.getTransportVersion().onOrAfter(TransportVersions.ZDT_NANOS_SUPPORT)) {
+                    o.writeZLong(instant.getEpochSecond());
+                } else {
+                    o.writeVLong(instant.getEpochSecond());
+                }
                 o.writeInt(instant.getNano());
             } else {
                 o.writeLong(instant.toEpochMilli());

server/src/test/java/org/elasticsearch/common/io/stream/AbstractStreamTests.java

@@ -10,7 +10,6 @@
 
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.TransportVersion;
-import org.elasticsearch.TransportVersions;
 import org.elasticsearch.common.CheckedBiConsumer;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -53,6 +52,8 @@
 
 import static java.time.Instant.ofEpochSecond;
 import static java.time.ZonedDateTime.ofInstant;
+import static org.elasticsearch.TransportVersions.ZDT_NANOS_SUPPORT;
+import static org.elasticsearch.TransportVersions.ZDT_NANOS_SUPPORT_BROKEN;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasToString;
@@ -726,26 +727,34 @@ public void testReadAfterReachingEndOfStream() throws IOException {
     }
 
     public void testZonedDateTimeSerialization() throws IOException {
-        checkZonedDateTimeSerialization(TransportVersions.ZDT_NANOS_SUPPORT);
+        checkZonedDateTimeSerialization(ZDT_NANOS_SUPPORT);
+    }
+
+    public void testZonedDateTimeMillisBwcSerializationV1() throws IOException {
+        checkZonedDateTimeSerialization(TransportVersionUtils.getPreviousVersion(ZDT_NANOS_SUPPORT_BROKEN));
     }
 
     public void testZonedDateTimeMillisBwcSerialization() throws IOException {
-        checkZonedDateTimeSerialization(TransportVersionUtils.getPreviousVersion(TransportVersions.ZDT_NANOS_SUPPORT));
+        checkZonedDateTimeSerialization(TransportVersionUtils.getPreviousVersion(ZDT_NANOS_SUPPORT));
     }
 
     public void checkZonedDateTimeSerialization(TransportVersion tv) throws IOException {
         assertGenericRoundtrip(ofInstant(Instant.EPOCH, randomZone()), tv);
         assertGenericRoundtrip(ofInstant(ofEpochSecond(1), randomZone()), tv);
         // just want to test a large number that will use 5+ bytes
         long maxEpochSecond = Integer.MAX_VALUE;
+        long minEpochSecond = tv.between(ZDT_NANOS_SUPPORT_BROKEN, ZDT_NANOS_SUPPORT) ? 0 : Integer.MIN_VALUE;
         assertGenericRoundtrip(ofInstant(ofEpochSecond(maxEpochSecond), randomZone()), tv);
-        assertGenericRoundtrip(ofInstant(ofEpochSecond(randomLongBetween(0, maxEpochSecond)), randomZone()), tv);
-        assertGenericRoundtrip(ofInstant(ofEpochSecond(randomLongBetween(0, maxEpochSecond), 1_000_000), randomZone()), tv);
-        assertGenericRoundtrip(ofInstant(ofEpochSecond(randomLongBetween(0, maxEpochSecond), 999_000_000), randomZone()), tv);
-        if (tv.onOrAfter(TransportVersions.ZDT_NANOS_SUPPORT)) {
-            assertGenericRoundtrip(ofInstant(ofEpochSecond(randomLongBetween(0, maxEpochSecond), 999_999_999), randomZone()), tv);
+        assertGenericRoundtrip(ofInstant(ofEpochSecond(randomLongBetween(minEpochSecond, maxEpochSecond)), randomZone()), tv);
+        assertGenericRoundtrip(ofInstant(ofEpochSecond(randomLongBetween(minEpochSecond, maxEpochSecond), 1_000_000), randomZone()), tv);
+        assertGenericRoundtrip(ofInstant(ofEpochSecond(randomLongBetween(minEpochSecond, maxEpochSecond), 999_000_000), randomZone()), tv);
+        if (tv.onOrAfter(ZDT_NANOS_SUPPORT)) {
+            assertGenericRoundtrip(
+                ofInstant(ofEpochSecond(randomLongBetween(minEpochSecond, maxEpochSecond), 999_999_999), randomZone()),
+                tv
+            );
             assertGenericRoundtrip(
-                ofInstant(ofEpochSecond(randomLongBetween(0, maxEpochSecond), randomIntBetween(0, 999_999_999)), randomZone()),
+                ofInstant(ofEpochSecond(randomLongBetween(minEpochSecond, maxEpochSecond), randomIntBetween(0, 999_999_999)), randomZone()),
                 tv
             );
         }