[AMORO-4193][ams] Fix MysqlDataTruncation caused by negative finishTime in TableProcessMeta (#4192)

* Fix MysqlDataTruncation caused by negative finishTime in TableProcessMeta

When creating a new TableProcessMeta via createProcessMeta(), finishTime
was initialized to -1. Long2TsConverter only converted 0 to null, so
Timestamp(-1) was passed to MySQL, which is outside the valid TIMESTAMP
range (1970-01-01 00:00:01 UTC ~), causing MysqlDataTruncation.

- Long2TsConverter: convert parameter <= 0 to null instead of only 0
- TableProcessMeta: initialize finishTime to 0 in createProcessMeta()
  for consistency with the of() factory method

Signed-off-by: Jiwon Park <jpark92@outlook.kr>

* Add unit tests for Long2TsConverter

Signed-off-by: Jiwon Park <jpark92@outlook.kr>

---------

Signed-off-by: Jiwon Park <jpark92@outlook.kr>

Author: j1wonpark

amoro-ams/src/main/java/org/apache/amoro/server/persistence/converter/Long2TsConverter.java

@@ -35,7 +35,7 @@ public class Long2TsConverter extends LongTypeHandler {
   @Override
   public void setNonNullParameter(PreparedStatement ps, int i, Long parameter, JdbcType jdbcType)
       throws SQLException {
-    if (parameter == 0) {
+    if (parameter <= 0) {
       ps.setTimestamp(i, null);
     } else {
       ps.setTimestamp(i, new Timestamp(parameter));

amoro-ams/src/main/java/org/apache/amoro/server/process/TableProcessMeta.java

@@ -202,7 +202,7 @@ public static TableProcessMeta createProcessMeta(TableProcess process) {
     tableProcessMeta.setExecutionEngine(process.getExecutionEngine());
     tableProcessMeta.setRetryNumber(0);
     tableProcessMeta.setCreateTime(System.currentTimeMillis());
-    tableProcessMeta.setFinishTime(-1);
+    tableProcessMeta.setFinishTime(0);
     tableProcessMeta.setFailMessage("");
     tableProcessMeta.setProcessParameters(process.getProcessParameters());
     tableProcessMeta.setSummary(process.getSummary());

amoro-ams/src/test/java/org/apache/amoro/server/persistence/converter/TestLong2TsConverter.java

@@ -0,0 +1,59 @@
+/*
+ * 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.amoro.server.persistence.converter;
+
+import org.apache.ibatis.type.JdbcType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+
+public class TestLong2TsConverter {
+
+  private Long2TsConverter converter;
+  private PreparedStatement ps;
+
+  @BeforeEach
+  void setUp() {
+    converter = new Long2TsConverter();
+    ps = Mockito.mock(PreparedStatement.class);
+  }
+
+  @Test
+  void testSetNullWhenZero() throws SQLException {
+    converter.setNonNullParameter(ps, 1, 0L, JdbcType.TIMESTAMP);
+    Mockito.verify(ps).setTimestamp(1, null);
+  }
+
+  @Test
+  void testSetNullWhenNegative() throws SQLException {
+    converter.setNonNullParameter(ps, 1, -1L, JdbcType.TIMESTAMP);
+    Mockito.verify(ps).setTimestamp(1, null);
+  }
+
+  @Test
+  void testSetTimestampWhenPositive() throws SQLException {
+    long now = System.currentTimeMillis();
+    converter.setNonNullParameter(ps, 1, now, JdbcType.TIMESTAMP);
+    Mockito.verify(ps).setTimestamp(1, new Timestamp(now));
+  }
+}