XuQianJin-Stars commented on code in PR #3518:
URL: https://github.com/apache/fluss/pull/3518#discussion_r3465663479
##########
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();
Review Comment:
`compactionFuture.get()` has no timeout. If the compaction stalls (lock
waits, slow IO, classloader leak, etc.) the entire tiering job is blocked
indefinitely. Please add a configurable timeout, and on timeout `cancel(true)`
the future and surface the failure (or fall back to an empty result with a
metric increment).
Additionally, the surrounding `catch (Exception e)` at line 120 will catch
`InterruptedException` without restoring the interrupt flag. If `get()` is
interrupted, please call `Thread.currentThread().interrupt()` before wrapping
the cause in `IOException`.
##########
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)) {
Review Comment:
Hardcoded 30s await for executor termination. This should be configurable; a
long-running compaction may not stop in 30s and we currently only emit a WARN,
leaking the executor. Suggest reading from `tableConfig` (e.g.
`lake.hudi.tiering.compaction.shutdown-timeout`) with a sensible default, and
bubbling up a failure if the executor cannot be shut down in time.
--
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]
