danny0405 commented on code in PR #18776:
URL: https://github.com/apache/hudi/pull/18776#discussion_r4059110740
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieBinaryCopyHandle.java:
##########
@@ -118,14 +118,26 @@ public void write() {
log.info("Schema evolution enabled for binary copy: {}",
schemaEvolutionEnabled);
records = this.writer.binaryCopy(inputFiles,
Collections.singletonList(path), writeScheMessageType, schemaEvolutionEnabled);
} catch (IOException e) {
+ closeWriterAfterFailure(e);
Review Comment:
Renamed to `closeWriterQuietly` and switched its cleanup to the shared
`CloseableUtils.closeSuppressing` helper.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/util/AutoCloseableUtils.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.hudi.util;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * Utility methods for closing {@link AutoCloseable} resources.
+ */
+public final class AutoCloseableUtils {
Review Comment:
Yes. After rebasing, I replaced this duplicate utility with the existing
`CloseableUtils` in hudi-common and extended `TestCloseableUtils`. It covers
null/successful close, checked and runtime exceptions, errors, and
self-suppression. All 60 selected tests passed, including the create/append
cleanup and Parquet binary-copy tests.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/util/AutoCloseableUtils.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.hudi.util;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * Utility methods for closing {@link AutoCloseable} resources.
+ */
+public final class AutoCloseableUtils {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(AutoCloseableUtils.class);
+
+ private AutoCloseableUtils() {
+ }
+
+ public static void closeWithSuppressed(AutoCloseable closeable, Throwable
failure) throws IOException {
+ if (closeable == null) {
+ return;
+ }
+ try {
+ closeable.close();
+ } catch (IOException ioe) {
+ if (failure != null) {
+ failure.addSuppressed(ioe);
+ } else {
+ throw ioe;
+ }
+ } catch (RuntimeException re) {
+ if (failure != null) {
+ failure.addSuppressed(re);
+ } else {
+ throw re;
+ }
+ } catch (Exception e) {
+ if (failure != null) {
+ failure.addSuppressed(e);
+ } else {
+ throw new IOException("Failed to close resource", e);
+ }
+ }
+ }
+
+ public static void closeQuietlyWithSuppressed(AutoCloseable closeable,
Throwable failure) {
+ try {
+ closeWithSuppressed(closeable, failure);
+ } catch (IOException | RuntimeException e) {
Review Comment:
The duplicate utility has now been removed. These call sites use the
existing `CloseableUtils.closeSuppressing`, which catches `Throwable` and
attaches close failures to the original exception, so checked `Exception` is
covered as well. `TestCloseableUtils` explicitly tests checked exceptions,
runtime exceptions, and errors.
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataParquetOutputStreamWriter.java:
##########
@@ -67,7 +69,12 @@ protected WriteSupport getWriteSupport(Configuration conf) {
parquetWriterbuilder.withDictionaryEncoding(parquetConfig.isDictionaryEnabled());
parquetWriterbuilder.withWriterVersion(ParquetWriter.DEFAULT_WRITER_VERSION);
parquetWriterbuilder.withConf(parquetConfig.getStorageConf().unwrapAs(Configuration.class));
- this.writer = parquetWriterbuilder.build();
+ try {
+ this.writer = parquetWriterbuilder.build();
+ } catch (IOException | RuntimeException e) {
+ closeQuietly(outputStream);
Review Comment:
The Flink, Spark, and Hadoop stream writers all use the same existing
`FileIOUtils.closeQuietly(outputStream)` helper. I retained that shared name so
the three call sites stay consistent without adding a separate stream-specific
wrapper to each class.
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/HoodieSparkParquetStreamWriter.java:
##########
@@ -35,23 +35,30 @@
import java.io.IOException;
+import static org.apache.hudi.io.util.FileIOUtils.closeQuietly;
+
public class HoodieSparkParquetStreamWriter implements HoodieSparkFileWriter,
AutoCloseable {
private final ParquetWriter<InternalRow> writer;
private final HoodieRowParquetWriteSupport writeSupport;
public HoodieSparkParquetStreamWriter(FSDataOutputStream outputStream,
HoodieRowParquetConfig parquetConfig) throws IOException {
this.writeSupport = parquetConfig.getWriteSupport();
- this.writer = new Builder<>(new
OutputStreamBackedOutputFile(outputStream), writeSupport)
- .withWriteMode(ParquetFileWriter.Mode.CREATE)
- .withCompressionCodec(parquetConfig.getCompressionCodecName())
- .withRowGroupSize(parquetConfig.getBlockSize())
- .withPageSize(parquetConfig.getPageSize())
- .withDictionaryPageSize(parquetConfig.getPageSize())
- .withDictionaryEncoding(parquetConfig.isDictionaryEnabled())
- .withWriterVersion(ParquetWriter.DEFAULT_WRITER_VERSION)
- .withConf(parquetConfig.getHadoopConf())
- .build();
+ try {
+ this.writer = new Builder<>(new
OutputStreamBackedOutputFile(outputStream), writeSupport)
+ .withWriteMode(ParquetFileWriter.Mode.CREATE)
+ .withCompressionCodec(parquetConfig.getCompressionCodecName())
+ .withRowGroupSize(parquetConfig.getBlockSize())
+ .withPageSize(parquetConfig.getPageSize())
+ .withDictionaryPageSize(parquetConfig.getPageSize())
+ .withDictionaryEncoding(parquetConfig.isDictionaryEnabled())
+ .withWriterVersion(ParquetWriter.DEFAULT_WRITER_VERSION)
+ .withConf(parquetConfig.getHadoopConf())
+ .build();
+ } catch (IOException | RuntimeException e) {
+ closeQuietly(outputStream);
Review Comment:
This uses the same `FileIOUtils.closeQuietly(outputStream)` helper as the
Flink and Hadoop stream writers. I retained the common helper name consistently
across all three.
##########
hudi-common/src/main/java/org/apache/hudi/common/bootstrap/index/hfile/HFileBootstrapIndexWriter.java:
##########
@@ -175,27 +176,65 @@ private void commit() {
* Close Writer Handles.
*/
public void close() {
- try {
- if (!closed) {
- indexByPartitionWriter.close();
- indexByFileIdWriter.close();
- closed = true;
- }
- } catch (IOException ioe) {
- throw new HoodieIOException(ioe.getMessage(), ioe);
+ if (closed) {
+ return;
+ }
+ Exception failure = closeWriter(indexByPartitionWriter, null);
+ failure = closeWriter(indexByFileIdWriter, failure);
+ closed = true;
+ if (failure != null) {
+ throw new HoodieException(failure.getMessage(), failure);
}
}
@Override
public void begin() {
try {
HFileContext context = HFileContext.builder().build();
- OutputStream outputStreamForPartitionWriter =
metaClient.getStorage().create(indexByPartitionPath);
- this.indexByPartitionWriter = new HFileWriterImpl(context,
outputStreamForPartitionWriter);
- OutputStream outputStreamForFileIdWriter =
metaClient.getStorage().create(indexByFileIdPath);
- this.indexByFileIdWriter = new HFileWriterImpl(context,
outputStreamForFileIdWriter);
+ this.indexByPartitionWriter = createHFileWriter(indexByPartitionPath,
context);
+ this.indexByFileIdWriter = createHFileWriter(indexByFileIdPath, context);
} catch (IOException ioe) {
+ closeAfterFailedBegin(ioe);
throw new HoodieIOException(ioe.getMessage(), ioe);
+ } catch (RuntimeException re) {
+ closeAfterFailedBegin(re);
+ throw re;
+ }
+ }
+
+ private HFileWriter createHFileWriter(StoragePath path, HFileContext
context) throws IOException {
+ OutputStream outputStream = metaClient.getStorage().create(path);
+ HFileWriter writer = null;
+ try {
+ writer = new HFileWriterImpl(context, outputStream);
+ return writer;
+ } finally {
+ if (writer == null) {
+ closeQuietly(outputStream);
+ }
+ }
+ }
+
+ private Exception closeWriter(HFileWriter writer, Exception failure) {
Review Comment:
Renamed to `closeHFileWriter`, matching the `HFileWriter` type's spelling.
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieParquetStreamWriter.java:
##########
@@ -49,16 +51,21 @@ public class HoodieParquetStreamWriter implements
HoodieAvroFileWriter, AutoClos
public HoodieParquetStreamWriter(FSDataOutputStream outputStream,
HoodieParquetConfig<HoodieAvroWriteSupport>
parquetConfig) throws IOException {
this.writeSupport = parquetConfig.getWriteSupport();
- this.writer = new Builder<IndexedRecord>(new
OutputStreamBackedOutputFile(outputStream), writeSupport)
- .withWriteMode(ParquetFileWriter.Mode.CREATE)
- .withCompressionCodec(parquetConfig.getCompressionCodecName())
- .withRowGroupSize(parquetConfig.getBlockSize())
- .withPageSize(parquetConfig.getPageSize())
- .withDictionaryPageSize(parquetConfig.getPageSize())
- .withDictionaryEncoding(parquetConfig.isDictionaryEnabled())
- .withWriterVersion(ParquetWriter.DEFAULT_WRITER_VERSION)
- .withConf(parquetConfig.getStorageConf().unwrapAs(Configuration.class))
- .build();
+ try {
+ this.writer = new Builder<IndexedRecord>(new
OutputStreamBackedOutputFile(outputStream), writeSupport)
+ .withWriteMode(ParquetFileWriter.Mode.CREATE)
+ .withCompressionCodec(parquetConfig.getCompressionCodecName())
+ .withRowGroupSize(parquetConfig.getBlockSize())
+ .withPageSize(parquetConfig.getPageSize())
+ .withDictionaryPageSize(parquetConfig.getPageSize())
+ .withDictionaryEncoding(parquetConfig.isDictionaryEnabled())
+ .withWriterVersion(ParquetWriter.DEFAULT_WRITER_VERSION)
+
.withConf(parquetConfig.getStorageConf().unwrapAs(Configuration.class))
+ .build();
+ } catch (IOException | RuntimeException e) {
+ closeQuietly(outputStream);
Review Comment:
All three stream writers use the existing
`FileIOUtils.closeQuietly(outputStream)` helper. I retained that common name
across Hadoop, Spark, and Flink; there are no separate per-writer close
wrappers to keep in sync.
##########
hudi-common/src/main/java/org/apache/hudi/common/bootstrap/index/hfile/HFileBootstrapIndexWriter.java:
##########
@@ -175,27 +176,65 @@ private void commit() {
* Close Writer Handles.
*/
public void close() {
- try {
- if (!closed) {
- indexByPartitionWriter.close();
- indexByFileIdWriter.close();
- closed = true;
- }
- } catch (IOException ioe) {
- throw new HoodieIOException(ioe.getMessage(), ioe);
+ if (closed) {
+ return;
+ }
+ Exception failure = closeWriter(indexByPartitionWriter, null);
+ failure = closeWriter(indexByFileIdWriter, failure);
+ closed = true;
+ if (failure != null) {
+ throw new HoodieException(failure.getMessage(), failure);
}
}
@Override
public void begin() {
try {
HFileContext context = HFileContext.builder().build();
- OutputStream outputStreamForPartitionWriter =
metaClient.getStorage().create(indexByPartitionPath);
- this.indexByPartitionWriter = new HFileWriterImpl(context,
outputStreamForPartitionWriter);
- OutputStream outputStreamForFileIdWriter =
metaClient.getStorage().create(indexByFileIdPath);
- this.indexByFileIdWriter = new HFileWriterImpl(context,
outputStreamForFileIdWriter);
+ this.indexByPartitionWriter = createHFileWriter(indexByPartitionPath,
context);
+ this.indexByFileIdWriter = createHFileWriter(indexByFileIdPath, context);
} catch (IOException ioe) {
+ closeAfterFailedBegin(ioe);
throw new HoodieIOException(ioe.getMessage(), ioe);
+ } catch (RuntimeException re) {
+ closeAfterFailedBegin(re);
+ throw re;
+ }
+ }
+
+ private HFileWriter createHFileWriter(StoragePath path, HFileContext
context) throws IOException {
+ OutputStream outputStream = metaClient.getStorage().create(path);
+ HFileWriter writer = null;
+ try {
+ writer = new HFileWriterImpl(context, outputStream);
+ return writer;
+ } finally {
+ if (writer == null) {
+ closeQuietly(outputStream);
+ }
+ }
+ }
+
+ private Exception closeWriter(HFileWriter writer, Exception failure) {
+ if (writer == null) {
+ return failure;
+ }
+ try {
+ writer.close();
+ } catch (IOException | RuntimeException e) {
+ if (failure == null) {
+ return e;
+ }
+ failure.addSuppressed(e);
+ }
+ return failure;
Review Comment:
The owning `close()` now clears both `indexByPartitionWriter` and
`indexByFileIdWriter` after attempting both closes. Assigning null to the
helper's `writer` parameter would only clear the local parameter, so the
assignments are made on the actual fields instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
