XuQianJin-Stars commented on code in PR #3518:
URL: https://github.com/apache/fluss/pull/3518#discussion_r3465663490
##########
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())) {
Review Comment:
A second `HudiWriteTableInfo` (and therefore a second
`HoodieFlinkWriteClient`) is created here while `recordWriter` already holds
one. Two write clients against the same table can race on Hudi internal state
(transaction manager, embedded timeline server, metadata table cache). Please
confirm that Hudi supports this pattern, or share the existing `hudiTableInfo`
between writer and compaction service.
--
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]
