XuQianJin-Stars commented on code in PR #3518:
URL: https://github.com/apache/fluss/pull/3518#discussion_r3465663494
##########
fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeWriter.java:
##########
@@ -91,22 +112,98 @@ public void write(LogRecord record) throws IOException {
public HudiWriteResult complete() throws IOException {
try {
Map<String, List<WriteStatus>> writeStatuses =
recordWriter.complete();
- return HudiWriteResult.fromWriteStatuses(writeStatuses,
Collections.emptyMap());
+ Map<String, List<WriteStatus>> compactionWriteStatuses =
Collections.emptyMap();
+ if (compactionFuture != null) {
+ compactionWriteStatuses = compactionFuture.get();
+ }
+ return HudiWriteResult.fromWriteStatuses(writeStatuses,
compactionWriteStatuses);
} catch (Exception e) {
throw new IOException("Failed to complete Hudi write.", e);
}
}
@Override
public void close() throws IOException {
+ IOException failure = closeCompactionExecutor();
+ failure = close(failure, recordWriter, "Hudi record writer");
+ failure = close(failure, ckpMetadata, "Hudi checkpoint metadata");
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ @Nullable
+ private IOException closeCompactionExecutor() {
+ try {
+ if (compactionFuture != null && !compactionFuture.isDone()) {
+ compactionFuture.cancel(true);
+ }
+ if (compactionExecutor != null) {
+ compactionExecutor.shutdown();
+ if (!compactionExecutor.awaitTermination(30,
TimeUnit.SECONDS)) {
+ LOG.warn("Failed to close Hudi compaction executor.");
+ }
+ }
+ return null;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return new IOException("Interrupted while closing Hudi compaction
executor.", e);
+ } catch (Exception e) {
+ return new IOException("Failed to close Hudi compaction
executor.", e);
+ }
+ }
+
+ @Nullable
+ private static IOException close(
+ @Nullable IOException failure, AutoCloseable closeable, String
resourceName) {
try {
- recordWriter.close();
- ckpMetadata.close();
+ closeable.close();
+ return failure;
} catch (Exception e) {
- throw new IOException("Failed to close HudiLakeWriter.", e);
+ IOException closeException =
+ new IOException("Failed to close " + resourceName + ".",
e);
+ if (failure == null) {
+ return closeException;
+ }
+ failure.addSuppressed(closeException);
+ return failure;
}
}
+ private boolean shouldRunCompaction(WriterInitContext writerInitContext) {
+ return
writerInitContext.tableInfo().getTableConfig().isDataLakeAutoCompaction()
+ && hudiTableInfo.getTableType() ==
HoodieTableType.MERGE_ON_READ;
+ }
+
+ private CompletableFuture<Map<String, List<WriteStatus>>>
executeCompactionAsync(
+ HudiCatalogProvider hudiCatalogProvider, WriterInitContext
writerInitContext) {
+ return CompletableFuture.supplyAsync(
+ () -> {
+ try (HudiWriteTableInfo compactionTableInfo =
+ HudiWriteTableInfo.create(
+ hudiCatalogProvider,
writerInitContext.tablePath())) {
+ HudiCompactionService compactionService =
+ new HudiCompactionService(
+ compactionTableInfo,
+ writerInitContext.tableBucket(),
+ writerInitContext.partition());
+ List<String> instantTimes =
+
compactionService.getInflightCompactionInstantTimes();
+ List<Pair<String, HoodieCompactionPlan>>
compactionPlans =
+
compactionService.getCompactionPlans(instantTimes);
+ return
compactionService.executeCompaction(compactionPlans);
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to execute Hudi compaction for table
{}, bucket {}.",
+ writerInitContext.tablePath(),
+ writerInitContext.tableBucket(),
+ e);
Review Comment:
`catch (Exception e) -> return Collections.emptyMap()` swallows compaction
failures completely. The committer cannot tell whether 'no compaction stats'
means 'nothing to compact' or 'compaction failed silently'. Please attach the
failure to `HudiWriteResult` (e.g. a nullable `compactionFailure` field) so the
committer can decide whether to surface or suppress, and so a metric can count
it.
--
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]
