nsivabalan commented on code in PR #18776:
URL: https://github.com/apache/hudi/pull/18776#discussion_r3285477807
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/BaseCreateHandle.java:
##########
@@ -168,6 +170,11 @@ protected HoodieRecord<T> updateFileName(HoodieRecord<T>
record, HoodieSchema sc
return record.prependMetaFields(schema, targetSchema, metadataValues,
prop);
}
+ private void closeFileWriterQuietly(Throwable failure) {
Review Comment:
did you intentionally not call `markClosed()` here or its an oversight.
##########
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:
should we also add `Exception` here.
##########
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:
same here
##########
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:
how about `closeWriterQuietly` to be in line w/ other naming in this patch
##########
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:
lets standardize all naming.
`closeOutputStreamQuietly`
##########
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:
do we have UTs for this class?
##########
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:
lets get the naming consistent across all writers.
`closeFileOutputStreamQuietly`
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/parquet/io/HoodieParquetBinaryCopyBase.java:
##########
@@ -137,7 +138,13 @@ protected void initFileWriter(Path outPutFile,
CompressionCodecName newCodecName
ParquetFileWriter.Mode writerMode = ParquetFileWriter.Mode.CREATE;
writer = new ParquetFileWriter(HadoopOutputFile.fromPath(outPutFile,
conf), schema, writerMode, DEFAULT_BLOCK_SIZE, MAX_PADDING_SIZE_DEFAULT,
DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH,
DEFAULT_STATISTICS_TRUNCATE_LENGTH,
ParquetProperties.DEFAULT_PAGE_WRITE_CHECKSUM_ENABLED);
- writer.start();
+ try {
+ writer.start();
+ } catch (Exception e) {
+ closeParquetFileWriterQuietly(writer);
Review Comment:
shouldn't we set the `writer = null` w/n closeParquetFileWriterQuietly
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieAvroHFileWriter.java:
##########
@@ -105,8 +106,14 @@ public HoodieAvroHFileWriter(String instantTime,
StoragePath file, HoodieHFileCo
.build();
StorageConfiguration<Configuration> storageConf = new
HadoopStorageConfiguration(conf);
StoragePath filePath = new StoragePath(this.file.toUri());
- OutputStream outputStream = HoodieStorageUtils.getStorage(filePath,
storageConf).create(filePath);
- this.writer = new HFileWriterImpl(context, outputStream);
+ OutputStream outputStream = HoodieStorageUtils.getStorage(filePath,
storageConf).create(filePath);
+ try {
+ this.writer = new HFileWriterImpl(context, outputStream);
+ } finally {
+ if (this.writer == null) {
+ closeQuietly(outputStream);
Review Comment:
+1
##########
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);
throw new HoodieIOException(e.getMessage(), e);
+ } catch (RuntimeException e) {
+ closeWriterAfterFailure(e);
+ throw e;
} finally {
this.recordsWritten = records;
this.insertRecordsWritten = records;
}
log.info("Finish rewriting " + this.path + ". Using " + timer.endTimer() +
" mills");
}
+ private void closeWriterAfterFailure(Throwable failure) {
+ try {
+ this.writer.close();
Review Comment:
+1
##########
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);
throw new HoodieIOException(e.getMessage(), e);
+ } catch (RuntimeException e) {
+ closeWriterAfterFailure(e);
+ throw e;
} finally {
this.recordsWritten = records;
this.insertRecordsWritten = records;
}
log.info("Finish rewriting " + this.path + ". Using " + timer.endTimer() +
" mills");
}
+ private void closeWriterAfterFailure(Throwable failure) {
+ try {
Review Comment:
+1
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/parquet/io/HoodieParquetBinaryCopyBase.java:
##########
@@ -505,32 +541,66 @@ private void addNullColumn(ColumnDescriptor descriptor,
long totalChunkValues, E
new ColumnChunkPageWriteStore(compressor, newSchema,
props.getAllocator(), props.getColumnIndexTruncateLength(),
props.getPageWriteChecksumEnabled(), null, numBlocksRewritten);
ColumnWriteStore cStore = props.newColumnWriteStore(newSchema, cPageStore);
ColumnWriter cWriter = cStore.getColumnWriter(descriptor);
- int dMax = descriptor.getMaxDefinitionLevel();
-
- for (int i = 0; i < totalChunkValues; i++) {
- int rlvl = 0;
- int dlvl = 0;
- if (dlvl == dMax) {
- // since we checked ether optional or repeated, dlvl should be > 0
- if (dlvl == 0) {
- throw new IOException("definition level is detected to be 0 for
column " + Arrays.stream(descriptor.getPath()).collect(Collectors.joining("."))
+ " to be nullified");
- }
- // we just write one null for the whole list at the top level,
- // instead of nullify the elements in the list one by one
- if (rlvl == 0) {
- cWriter.writeNull(rlvl, dlvl - 1);
+ Throwable failure = null;
+ try {
+ int dMax = descriptor.getMaxDefinitionLevel();
+
+ for (int i = 0; i < totalChunkValues; i++) {
+ int rlvl = 0;
+ int dlvl = 0;
+ if (dlvl == dMax) {
+ // since we checked ether optional or repeated, dlvl should be > 0
+ if (dlvl == 0) {
+ throw new IOException("definition level is detected to be 0 for
column " + Arrays.stream(descriptor.getPath()).collect(Collectors.joining("."))
+ " to be nullified");
+ }
+ // we just write one null for the whole list at the top level,
+ // instead of nullify the elements in the list one by one
+ if (rlvl == 0) {
+ cWriter.writeNull(rlvl, dlvl - 1);
+ }
+ } else {
+ cWriter.writeNull(rlvl, dlvl); // 因为repeatition
level没有重复所以后面都是以0在第一层，definition level是字段path的第0层
Review Comment:
sorry. can we avoid comments in non english language. I understand it was
not from this patch.
but lets fix it.
##########
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:
`closeHfileWriter`
##########
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:
shouldn't we set the `writer = null` at the end of the method
##########
hudi-hadoop-common/src/main/java/org/apache/hudi/parquet/io/HoodieParquetBinaryCopyBase.java:
##########
@@ -505,32 +541,66 @@ private void addNullColumn(ColumnDescriptor descriptor,
long totalChunkValues, E
new ColumnChunkPageWriteStore(compressor, newSchema,
props.getAllocator(), props.getColumnIndexTruncateLength(),
props.getPageWriteChecksumEnabled(), null, numBlocksRewritten);
ColumnWriteStore cStore = props.newColumnWriteStore(newSchema, cPageStore);
ColumnWriter cWriter = cStore.getColumnWriter(descriptor);
- int dMax = descriptor.getMaxDefinitionLevel();
-
- for (int i = 0; i < totalChunkValues; i++) {
- int rlvl = 0;
- int dlvl = 0;
- if (dlvl == dMax) {
- // since we checked ether optional or repeated, dlvl should be > 0
- if (dlvl == 0) {
- throw new IOException("definition level is detected to be 0 for
column " + Arrays.stream(descriptor.getPath()).collect(Collectors.joining("."))
+ " to be nullified");
- }
- // we just write one null for the whole list at the top level,
- // instead of nullify the elements in the list one by one
- if (rlvl == 0) {
- cWriter.writeNull(rlvl, dlvl - 1);
+ Throwable failure = null;
+ try {
+ int dMax = descriptor.getMaxDefinitionLevel();
+
+ for (int i = 0; i < totalChunkValues; i++) {
+ int rlvl = 0;
+ int dlvl = 0;
+ if (dlvl == dMax) {
+ // since we checked ether optional or repeated, dlvl should be > 0
+ if (dlvl == 0) {
+ throw new IOException("definition level is detected to be 0 for
column " + Arrays.stream(descriptor.getPath()).collect(Collectors.joining("."))
+ " to be nullified");
+ }
+ // we just write one null for the whole list at the top level,
+ // instead of nullify the elements in the list one by one
+ if (rlvl == 0) {
+ cWriter.writeNull(rlvl, dlvl - 1);
+ }
+ } else {
+ cWriter.writeNull(rlvl, dlvl); // 因为repeatition
level没有重复所以后面都是以0在第一层，definition level是字段path的第0层
}
- } else {
- cWriter.writeNull(rlvl, dlvl); // 因为repeatition
level没有重复所以后面都是以0在第一层，definition level是字段path的第0层
+ cStore.endRecord();
}
- cStore.endRecord();
- }
- cStore.flush();
- cPageStore.flushToFileWriter(writer);
+ cStore.flush();
+ cPageStore.flushToFileWriter(writer);
+ } catch (IOException | RuntimeException e) {
+ failure = e;
+ throw e;
+ } finally {
+ closeColumnWriters(cStore, cWriter, failure);
+ }
+ }
- cStore.close();
- cWriter.close();
+ private void closeColumnWriters(ColumnWriteStore cStore, ColumnWriter
cWriter, Throwable failure) {
+ RuntimeException closeException = null;
+ try {
+ if (cStore != null) {
+ cStore.close();
+ }
+ } catch (RuntimeException re) {
+ closeException = re;
+ }
+ try {
+ if (cWriter != null) {
+ cWriter.close();
+ }
+ } catch (RuntimeException re) {
+ if (closeException == null) {
+ closeException = re;
+ } else {
+ closeException.addSuppressed(re);
+ }
+ }
+ if (closeException != null) {
+ if (failure != null) {
+ failure.addSuppressed(closeException);
+ } else {
+ throw closeException;
+ }
+ }
Review Comment:
should we set `cWriter = null` in the end?
--
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]
