[kafka] Preserve unknown JSON fields in rescue column

fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java

@@ -19,8 +19,10 @@
 
 import org.apache.fluss.client.Connection;
 import org.apache.fluss.client.ConnectionFactory;
+import org.apache.fluss.client.FlussConnection;
 import org.apache.fluss.client.metadata.KvSnapshotMetadata;
 import org.apache.fluss.client.metadata.KvSnapshots;
+import org.apache.fluss.client.metadata.MetadataUpdater;
 import org.apache.fluss.client.metadata.TestingMetadataUpdater;
 import org.apache.fluss.client.table.Table;
 import org.apache.fluss.client.table.writer.AppendWriter;
@@ -1211,19 +1213,24 @@ void testKvSnapshotLeaseAfterCoordinatorServerRestart() throws Exception {
         assertThat(zkClient.getKvSnapshotLeaseMetadata(lease.leaseId())).isPresent();
 
         restartCoordinatorServer(zkClient);
-        FLUSS_CLUSTER_EXTENSION.waitUntilAllGatewayHasSameMetadata();
-
         lease.dropLease().get();
         assertThat(zkClient.getKvSnapshotLeaseMetadata(lease.leaseId())).isNotPresent();
     }
 
     private void restartCoordinatorServer(ZooKeeperClient zkClient) throws Exception {
+        MetadataUpdater metadataUpdater = ((FlussConnection) conn).getMetadataUpdater();
+        ServerNode cachedCoordinator = metadataUpdater.getCoordinatorServer();
         FLUSS_CLUSTER_EXTENSION.stopCoordinatorServer();
         waitUntil(
                 () -> !zkClient.getCoordinatorLeaderAddress().isPresent(),
                 Duration.ofMinutes(1),
                 "Coordinator server node still exists in ZooKeeper");
         FLUSS_CLUSTER_EXTENSION.startCoordinatorServer();
+        // Election completes before the new coordinator address reaches every tablet server.
+        // Wait for server metadata to converge so that the client's retry can discover it.
+        FLUSS_CLUSTER_EXTENSION.waitUntilAllGatewayHasSameMetadata();
+        // Keep the client cache stale to exercise the lease's metadata refresh and retry.
+        assertThat(metadataUpdater.getCoordinatorServer()).isEqualTo(cachedCoordinator);
     }
 
     @Test

fluss-common/src/main/java/org/apache/fluss/row/TimestampNtz.java

@@ -137,12 +137,8 @@ public long toEpochMicros() {
 
     /** Converts this {@link TimestampNtz} object to a {@link LocalDateTime}. */
     public LocalDateTime toLocalDateTime() {
-        int date = (int) (millisecond / MILLIS_PER_DAY);
-        long time = millisecond % MILLIS_PER_DAY;
-        if (time < 0) {
-            --date;
-            time += MILLIS_PER_DAY;
-        }
+        long date = Math.floorDiv(millisecond, MILLIS_PER_DAY);
+        long time = Math.floorMod(millisecond, MILLIS_PER_DAY);
         long nanoOfDay = time * 1_000_000L + nanoOfMillisecond;
         LocalDate localDate = LocalDate.ofEpochDay(date);
         LocalTime localTime = LocalTime.ofNanoOfDay(nanoOfDay);

fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampLtzColumnVector.java

@@ -57,10 +57,12 @@ public TimestampLtz getTimestampLtz(int i, int precision) {
             return TimestampLtz.fromEpochMillis(((TimeStampMilliVector) valueVector).get(i));
         } else if (valueVector instanceof TimeStampMicroVector) {
             long micros = ((TimeStampMicroVector) valueVector).get(i);
-            return TimestampLtz.fromEpochMillis(micros / 1000, (int) (micros % 1000) * 1000);
+            return TimestampLtz.fromEpochMillis(
+                    Math.floorDiv(micros, 1_000L), (int) Math.floorMod(micros, 1_000L) * 1_000);
         } else {
             long nanos = ((TimeStampNanoVector) valueVector).get(i);
-            return TimestampLtz.fromEpochMillis(nanos / 1_000_000, (int) (nanos % 1_000_000));
+            return TimestampLtz.fromEpochMillis(
+                    Math.floorDiv(nanos, 1_000_000L), (int) Math.floorMod(nanos, 1_000_000L));
         }
     }
 

fluss-common/src/main/java/org/apache/fluss/row/arrow/vectors/ArrowTimestampNtzColumnVector.java

@@ -56,10 +56,12 @@ public TimestampNtz getTimestampNtz(int i, int precision) {
             return TimestampNtz.fromMillis(((TimeStampMilliVector) valueVector).get(i));
         } else if (valueVector instanceof TimeStampMicroVector) {
             long micros = ((TimeStampMicroVector) valueVector).get(i);
-            return TimestampNtz.fromMillis(micros / 1000, (int) (micros % 1000) * 1000);
+            return TimestampNtz.fromMillis(
+                    Math.floorDiv(micros, 1_000L), (int) Math.floorMod(micros, 1_000L) * 1_000);
         } else {
             long nanos = ((TimeStampNanoVector) valueVector).get(i);
-            return TimestampNtz.fromMillis(nanos / 1_000_000, (int) (nanos % 1_000_000));
+            return TimestampNtz.fromMillis(
+                    Math.floorDiv(nanos, 1_000_000L), (int) Math.floorMod(nanos, 1_000_000L));
         }
     }
 

fluss-common/src/main/java/org/apache/fluss/row/arrow/writers/ArrowTimestampLtzWriter.java

@@ -27,6 +27,8 @@
 import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.TimeStampVector;
 import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.types.pojo.ArrowType;
 
+import static org.apache.fluss.utils.DateTimeUtils.timestampToMicros;
+import static org.apache.fluss.utils.DateTimeUtils.timestampToNanos;
 import static org.apache.fluss.utils.Preconditions.checkState;
 
 /** {@link ArrowFieldWriter} for TimestampLtz. */
@@ -64,16 +66,17 @@ public void doWrite(int rowIndex, DataGetters row, int ordinal, boolean handleSa
             }
         } else if (vector instanceof TimeStampMicroVector) {
             long microSec =
-                    timestamp.getEpochMillisecond() * 1000
-                            + timestamp.getNanoOfMillisecond() / 1000;
+                    timestampToMicros(
+                            timestamp.getEpochMillisecond(), timestamp.getNanoOfMillisecond());
             if (handleSafe) {
                 vector.setSafe(rowIndex, microSec);
             } else {
                 vector.set(rowIndex, microSec);
             }
         } else {
             long nanoSec =
-                    timestamp.getEpochMillisecond() * 1_000_000 + timestamp.getNanoOfMillisecond();
+                    timestampToNanos(
+                            timestamp.getEpochMillisecond(), timestamp.getNanoOfMillisecond());
             if (handleSafe) {
                 vector.setSafe(rowIndex, nanoSec);
             } else {

fluss-common/src/main/java/org/apache/fluss/row/arrow/writers/ArrowTimestampNtzWriter.java

@@ -27,6 +27,8 @@
 import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.TimeStampVector;
 import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.types.pojo.ArrowType;
 
+import static org.apache.fluss.utils.DateTimeUtils.timestampToMicros;
+import static org.apache.fluss.utils.DateTimeUtils.timestampToNanos;
 import static org.apache.fluss.utils.Preconditions.checkState;
 
 /** {@link ArrowFieldWriter} for TimestampNtz. */
@@ -64,15 +66,15 @@ public void doWrite(int rowIndex, DataGetters row, int ordinal, boolean handleSa
             }
         } else if (vector instanceof TimeStampMicroVector) {
             long microSec =
-                    timestamp.getMillisecond() * 1000 + timestamp.getNanoOfMillisecond() / 1000;
+                    timestampToMicros(timestamp.getMillisecond(), timestamp.getNanoOfMillisecond());
             if (handleSafe) {
                 vector.setSafe(rowIndex, microSec);
             } else {
                 vector.set(rowIndex, microSec);
             }
         } else {
             long nanoSec =
-                    timestamp.getMillisecond() * 1_000_000 + timestamp.getNanoOfMillisecond();
+                    timestampToNanos(timestamp.getMillisecond(), timestamp.getNanoOfMillisecond());
             if (handleSafe) {
                 vector.setSafe(rowIndex, nanoSec);
             } else {

fluss-common/src/main/java/org/apache/fluss/utils/DateTimeUtils.java

@@ -71,6 +71,35 @@ public class DateTimeUtils {
                     .optionalEnd()
                     .toFormatter();
 
+    /**
+     * Converts a timestamp's milliseconds and nanoseconds within the millisecond to microseconds.
+     * Sub-microsecond digits are discarded.
+     *
+     * @throws ArithmeticException if the result does not fit in a long
+     */
+    public static long timestampToMicros(long millisecond, int nanoOfMillisecond) {
+        return scaleTimestamp(millisecond, MICROS_PER_MILLIS, nanoOfMillisecond / NANOS_PER_MICROS);
+    }
+
+    /**
+     * Converts a timestamp's milliseconds and nanoseconds within the millisecond to nanoseconds.
+     *
+     * @throws ArithmeticException if the result does not fit in a long
+     */
+    public static long timestampToNanos(long millisecond, int nanoOfMillisecond) {
+        return scaleTimestamp(millisecond, 1_000_000L, nanoOfMillisecond);
+    }
+
+    private static long scaleTimestamp(long millisecond, long unitsPerMillis, long fraction) {
+        if (millisecond < 0) {
+            // The non-negative fraction can bring the final value back into range even when
+            // multiplying the floor-rounded millisecond would underflow at Long.MIN_VALUE.
+            return Math.addExact(
+                    Math.multiplyExact(millisecond + 1, unitsPerMillis), fraction - unitsPerMillis);
+        }
+        return Math.addExact(Math.multiplyExact(millisecond, unitsPerMillis), fraction);
+    }
+
     public static LocalDate toLocalDate(int date) {
         return julianToLocalDate(date + EPOCH_JULIAN);
     }

fluss-common/src/test/java/org/apache/fluss/row/TimestampNtzTest.java

@@ -19,7 +19,9 @@
 
 import org.junit.jupiter.api.Test;
 
+import java.time.Instant;
 import java.time.LocalDateTime;
+import java.time.ZoneOffset;
 import java.util.TimeZone;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -46,6 +48,29 @@ public void testNormal() {
         assertThat(TimestampNtz.fromLocalDateTime(ldt3).toLocalDateTime()).isEqualTo(ldt3);
     }
 
+    @Test
+    void testToLocalDateTimeAcrossMillisecondRange() {
+        long[] milliseconds = {
+            Long.MIN_VALUE,
+            (long) Integer.MIN_VALUE * 86_400_000 - 1,
+            -86_400_001,
+            -1,
+            0,
+            86_400_000,
+            ((long) Integer.MAX_VALUE + 1) * 86_400_000,
+            Long.MAX_VALUE
+        };
+        for (long millis : milliseconds) {
+            for (int nanos : new int[] {0, 999_999}) {
+                LocalDateTime expected =
+                        LocalDateTime.ofInstant(
+                                Instant.ofEpochMilli(millis).plusNanos(nanos), ZoneOffset.UTC);
+                assertThat(TimestampNtz.fromMillis(millis, nanos).toLocalDateTime())
+                        .isEqualTo(expected);
+            }
+        }
+    }
+
     @Test
     public void testDaylightSavingTime() {
         TimeZone tz = TimeZone.getDefault();

fluss-common/src/test/java/org/apache/fluss/row/arrow/ArrowTimestampWriterTest.java

@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.row.arrow;
+
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+import org.apache.fluss.row.arrow.writers.ArrowFieldWriter;
+import org.apache.fluss.row.columnar.ColumnVector;
+import org.apache.fluss.row.columnar.ColumnarRow;
+import org.apache.fluss.shaded.arrow.org.apache.arrow.memory.RootAllocator;
+import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.TimeStampVector;
+import org.apache.fluss.shaded.arrow.org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.types.RowType;
+import org.apache.fluss.utils.ArrowUtils;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Verifies timestamp range checks and round trips through Arrow vectors. */
+class ArrowTimestampWriterTest {
+
+    @ParameterizedTest
+    @MethodSource("timestampWriters")
+    void testRoundTripsTimestampBoundaries(int precision, boolean withZone, boolean handleSafe) {
+        DataType type =
+                withZone ? DataTypes.TIMESTAMP_LTZ(precision) : DataTypes.TIMESTAMP(precision);
+        RowType rowType = DataTypes.ROW(type);
+        long unitsPerMilli = precision == 6 ? 1_000 : 1_000_000;
+        int nanosPerUnit = precision == 6 ? 1_000 : 1;
+        try (RootAllocator allocator = new RootAllocator(1024 * 1024);
+                VectorSchemaRoot root =
+                        VectorSchemaRoot.create(ArrowUtils.toArrowSchema(rowType), allocator)) {
+            root.allocateNew();
+            TimeStampVector vector = (TimeStampVector) root.getVector(0);
+            ArrowFieldWriter writer = ArrowUtils.createArrowFieldWriter(vector, type);
+            long[] values = {
+                Long.MIN_VALUE,
+                -1_000_001,
+                -unitsPerMilli,
+                -unitsPerMilli + 1,
+                -1,
+                0,
+                1,
+                1_000_001,
+                Long.MAX_VALUE
+            };
+            for (int i = 0; i < values.length; i++) {
+                long millis = Math.floorDiv(values[i], unitsPerMilli);
+                int nanos = (int) Math.floorMod(values[i], unitsPerMilli) * nanosPerUnit;
+                writer.write(i, timestampRow(millis, nanos, withZone), 0, handleSafe);
+                assertThat(vector.get(i)).isEqualTo(values[i]);
+            }
+            root.setRowCount(values.length);
+            ArrowReader reader =
+                    new ArrowReader(
+                            new ColumnVector[] {ArrowUtils.createArrowColumnVector(vector, type)},
+                            root.getRowCount());
+            for (int i = 0; i < values.length; i++) {
+                long millis = Math.floorDiv(values[i], unitsPerMilli);
+                int nanos = (int) Math.floorMod(values[i], unitsPerMilli) * nanosPerUnit;
+                ColumnarRow row = reader.read(i);
+                if (withZone) {
+                    assertThat(row.getTimestampLtz(0, precision))
+                            .isEqualTo(TimestampLtz.fromEpochMillis(millis, nanos));
+                } else {
+                    assertThat(row.getTimestampNtz(0, precision))
+                            .isEqualTo(TimestampNtz.fromMillis(millis, nanos));
+                }
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @MethodSource("timestampWriters")
+    void testRejectsOverflowBeforeChangingVector(
+            int precision, boolean withZone, boolean handleSafe) {
+        DataType type =
+                withZone ? DataTypes.TIMESTAMP_LTZ(precision) : DataTypes.TIMESTAMP(precision);
+        RowType rowType = DataTypes.ROW(type);
+        long unitsPerMilli = precision == 6 ? 1_000 : 1_000_000;
+        int nanosPerUnit = precision == 6 ? 1_000 : 1;
+        try (RootAllocator allocator = new RootAllocator(1024 * 1024);
+                VectorSchemaRoot root =
+                        VectorSchemaRoot.create(ArrowUtils.toArrowSchema(rowType), allocator)) {
+            root.allocateNew();
+            TimeStampVector vector = (TimeStampVector) root.getVector(0);
+            ArrowFieldWriter writer = ArrowUtils.createArrowFieldWriter(vector, type);
+            vector.set(0, 42);
+            for (long limit : new long[] {Long.MIN_VALUE, Long.MAX_VALUE}) {
+                long millis = Math.floorDiv(limit, unitsPerMilli);
+                int nanos = (int) Math.floorMod(limit, unitsPerMilli) * nanosPerUnit;
+                int outsideNanos = nanos + (limit < 0 ? -nanosPerUnit : nanosPerUnit);
+                assertThatThrownBy(
+                                () ->
+                                        writer.write(
+                                                0,
+                                                timestampRow(millis, outsideNanos, withZone),
+                                                0,
+                                                handleSafe))
+                        .isInstanceOf(ArithmeticException.class);
+                long outsideMillis = millis + (limit < 0 ? -1 : 1);
+                assertThatThrownBy(
+                                () ->
+                                        writer.write(
+                                                0,
+                                                timestampRow(outsideMillis, nanos, withZone),
+                                                0,
+                                                handleSafe))
+                        .isInstanceOf(ArithmeticException.class);
+                assertThat(vector.get(0)).isEqualTo(42);
+            }
+        }
+    }
+
+    private static GenericRow timestampRow(long millis, int nanos, boolean withZone) {
+        return GenericRow.of(
+                withZone
+                        ? TimestampLtz.fromEpochMillis(millis, nanos)
+                        : TimestampNtz.fromMillis(millis, nanos));
+    }
+
+    private static Stream<Arguments> timestampWriters() {
+        return Stream.of(
+                Arguments.of(6, false, false), Arguments.of(6, false, true),
+                Arguments.of(6, true, false), Arguments.of(6, true, true),
+                Arguments.of(9, false, false), Arguments.of(9, false, true),
+                Arguments.of(9, true, false), Arguments.of(9, true, true));
+    }
+}

fluss-kafka/pom.xml

@@ -63,7 +63,18 @@
             <version>${project.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.fluss</groupId>
+            <artifactId>fluss-shaded-jackson</artifactId>
+        </dependency>
+
         <!-- test dependency -->
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fluss-client</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.curator</groupId>
             <artifactId>curator-test</artifactId>