[hive] Bug fix for stuck thread on hive prepareCommit failure

paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/mapred/PaimonOutputCommitter.java

@@ -271,9 +271,20 @@ private static void createPreCommitFile(
     }
 
     private static List<CommitMessage> readPreCommitFile(Path location, FileIO io) {
-        try (ObjectInputStream objectInputStream =
-                new ObjectInputStream(io.newInputStream(location))) {
-            return (List<CommitMessage>) objectInputStream.readObject();
+        try {
+            if (!io.exists(location)) {
+                LOG.warn(
+                        "preCommit file {} was not generated. The task did not "
+                                + "reach prepareCommit, so there are no commit "
+                                + "messages to abort. Skipping commit cleanup "
+                                + "for this task slot.",
+                        location);
+                return Collections.emptyList();
+            }
+            try (ObjectInputStream objectInputStream =
+                    new ObjectInputStream(io.newInputStream(location))) {
+                return (List<CommitMessage>) objectInputStream.readObject();
+            }
         } catch (ClassNotFoundException | IOException e) {
             throw new RuntimeException(
                     String.format("Can not read or parse CommitMessage file: %s", location));

paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/mapred/PaimonOutputCommitterTest.java

@@ -0,0 +1,90 @@
+/*
+ * 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.paimon.hive.mapred;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.table.sink.CommitMessage;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.ObjectOutputStream;
+import java.lang.reflect.Method;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link PaimonOutputCommitter}. */
+public class PaimonOutputCommitterTest {
+
+    @Rule public TemporaryFolder folder = new TemporaryFolder();
+
+    @Test
+    public void readPreCommitFileReturnsEmptyWhenFileMissing() throws Exception {
+        Path missing = new Path(folder.newFolder().getAbsolutePath(), "task_0.preCommit");
+        List<CommitMessage> result = invokeReadPreCommitFile(missing);
+        assertThat(result).isEmpty();
+    }
+
+    @Test
+    public void readPreCommitFileThrowsOnCorruptFile() throws Exception {
+        java.io.File f = folder.newFile("task_0.preCommit");
+        try (java.io.FileOutputStream fos = new java.io.FileOutputStream(f)) {
+            fos.write(new byte[] {0x00, 0x01, 0x02, 0x03});
+        }
+        Path corrupt = new Path(f.getAbsolutePath());
+        org.assertj.core.api.Assertions.assertThatThrownBy(() -> invokeReadPreCommitFile(corrupt))
+                .isInstanceOf(RuntimeException.class)
+                .hasMessageContaining("Can not read or parse CommitMessage file");
+    }
+
+    @Test
+    public void readPreCommitFileReturnsListForValidFile() throws Exception {
+        java.io.File f = folder.newFile("task_0.preCommit");
+        try (ObjectOutputStream oos = new ObjectOutputStream(new java.io.FileOutputStream(f))) {
+            oos.writeObject(Collections.<CommitMessage>emptyList());
+        }
+        Path written = new Path(f.getAbsolutePath());
+        List<CommitMessage> result = invokeReadPreCommitFile(written);
+        assertThat(result).isNotNull().isEmpty();
+    }
+
+    @SuppressWarnings("unchecked")
+    private static List<CommitMessage> invokeReadPreCommitFile(Path location) throws Exception {
+        Method m =
+                PaimonOutputCommitter.class.getDeclaredMethod(
+                        "readPreCommitFile", Path.class, org.apache.paimon.fs.FileIO.class);
+        m.setAccessible(true);
+        try {
+            return (List<CommitMessage>) m.invoke(null, location, LocalFileIO.create());
+        } catch (java.lang.reflect.InvocationTargetException e) {
+            Throwable cause = e.getCause();
+            if (cause instanceof RuntimeException) {
+                throw (RuntimeException) cause;
+            }
+            if (cause instanceof Error) {
+                throw (Error) cause;
+            }
+            throw new RuntimeException(cause);
+        }
+    }
+}