fix(hdfs): Nullify hdfsFile_ before checking success status (facebookincubator#14666)

Summary:
driver_->CloseFile can fail if the task is interrupted. In that case, hdfsFile_ is not reset to nullptr. Later, when the HdfsWriteFile destructor runs, it invokes close() again, resulting in a double close. This leads to a fatal JNI error:

```java
25/08/25 08:12:06 INFO [dispatcher-Executor] Executor: Executor is trying to kill task 43.2 in stage 11.0 (TID 617), reason: another attempt succeeded
HdfsWriteFile.cpp:57, Function:close, Expression: success == 0 (-1 vs. 0) Failed to close hdfs file: IOException: Failed to shutdown streamer, Source: RUNTIME, ErrorCode: INVALID_STATE
```
Eventually, the JVM crashes with a segmentation fault during a JNI call to FSDataOutputStream.close():

```java
#
# A fatal error has been detected by the Java Runtime Environment:
#
#  SIGSEGV (0xb) at pc=0x00007fa2ec7e0c48, pid=174755, tid=0x00007fa220262700
#
# JRE version: OpenJDK Runtime Environment (8.0_252-b09) (build 1.8.0_252-b09)
# Java VM: OpenJDK 64-Bit Server VM (25.252-b09 mixed mode linux-amd64 compressed oops)
# Problematic frame:
# V  [libjvm.so+0x68fc48]  jni_invoke_nonstatic(JNIEnv_*, JavaValue*, _jobject*, JNICallType, _jmethodID*, JNI_ArgumentPusher*, Thread*)+0x48
#
```

Stack trace excerpt showing the duplicate close() call path:
```c++
facebookincubator#8  invokeMethodOnJclass(..., "org/apache/hadoop/fs/FSDataOutputStream", "close", "()V")
facebookincubator#10 hdfsCloseFile(...)
facebookincubator#11 facebook::velox::HdfsWriteFile::close()
facebookincubator#12 facebook::velox::HdfsWriteFile::~HdfsWriteFile()
facebookincubator#14 facebook::velox::dwio::common::WriteFileSink::~WriteFileSink()
facebookincubator#15 facebook::velox::parquet::Writer::abort()
```

Pull Request resolved: facebookincubator#14666

Reviewed By: gggrace14

Differential Revision: D82880508

Pulled By: xiaoxmeng

fbshipit-source-id: c49467744b7784c3373faf61e8a46058f5243caf

Author: boneanxs

velox/connectors/hive/storage_adapters/hdfs/HdfsWriteFile.cpp

@@ -54,12 +54,14 @@ HdfsWriteFile::~HdfsWriteFile() {
 
 void HdfsWriteFile::close() {
   int success = driver_->CloseFile(hdfsClient_, hdfsFile_);
+  common::testutil::TestValue::adjust(
+      "facebook::velox::connectors::hive::HdfsWriteFile::close", &success);
+  hdfsFile_ = nullptr;
   VELOX_CHECK_EQ(
       success,
       0,
       "Failed to close hdfs file: {}",
       driver_->GetLastExceptionRootCause());
-  hdfsFile_ = nullptr;
 }
 
 void HdfsWriteFile::flush() {

velox/connectors/hive/storage_adapters/hdfs/tests/HdfsFileSystemTest.cpp

@@ -21,6 +21,7 @@
 #include "gtest/gtest.h"
 #include "velox/common/base/Exceptions.h"
 #include "velox/common/base/tests/GTestUtils.h"
+#include "velox/common/testutil/TestValue.h"
 #include "velox/connectors/hive/storage_adapters/hdfs/HdfsReadFile.h"
 #include "velox/connectors/hive/storage_adapters/hdfs/RegisterHdfsFileSystem.h"
 #include "velox/connectors/hive/storage_adapters/hdfs/tests/HdfsMiniCluster.h"
@@ -525,3 +526,34 @@ TEST_F(HdfsFileSystemTest, readFailures) {
       std::string(miniCluster->nameNodePort()));
   verifyFailures(driver, hdfs);
 }
+
+DEBUG_ONLY_TEST_F(HdfsFileSystemTest, writeFilePreventsDoubleClose) {
+  common::testutil::TestValue::enable();
+
+  int closeCallCount = 0;
+
+  SCOPED_TESTVALUE_SET(
+      "facebook::velox::connectors::hive::HdfsWriteFile::close",
+      std::function<void(int*)>([&closeCallCount](int* success) {
+        ++closeCallCount;
+        if (closeCallCount == 1) {
+          *success = -1;
+        }
+      }));
+
+  auto writeFile = openFileForWrite("/test_double_close.txt");
+
+  writeFile->append("test data");
+  writeFile->flush();
+
+  VELOX_ASSERT_THROW(writeFile->close(), "Failed to close hdfs file:");
+
+  EXPECT_EQ(closeCallCount, 1);
+
+  // Destructor should not call close() again because hdfsFile_ is nullptr
+  // The closeCallCount should remain 1.
+  writeFile.reset();
+  EXPECT_EQ(closeCallCount, 1);
+
+  common::testutil::TestValue::disable();
+}