XuQianJin-Stars commented on code in PR #3518:
URL: https://github.com/apache/fluss/pull/3518#discussion_r3465663462
##########
fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeCommitter.java:
##########
@@ -207,6 +211,53 @@ public void close() throws Exception {
IOUtils.closeQuietly(hudiTableInfo, "hudi table info");
}
+ @Nullable
+ private String commitCompactionAndSchedule(
+ Map<String, HudiWriteStats> compactionWriteStats, Map<String,
String> commitMetadata) {
+ if (!isAutoCompactionEnabled()) {
+ return null;
+ }
+
+ String latestCommittedInstant = null;
+ try {
+ latestCommittedInstant =
+ compactionService.commitCompaction(compactionWriteStats,
commitMetadata);
+ if (latestCommittedInstant != null) {
+ LOG.info("Committed Hudi compaction instant {}.",
latestCommittedInstant);
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to commit Hudi compaction for table {}. "
+ + "The next tiering round can rollback and retry
pending compactions.",
+ hudiTableInfo.getTablePath(),
+ e);
+ }
+
+ try {
+ compactionService.scheduleCompaction();
+ compactionService.markSelectedCompactionsInflight();
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to schedule Hudi compaction for table {}.",
+ hudiTableInfo.getTablePath(),
+ e);
+ }
+ return latestCommittedInstant;
+ }
+
+ private boolean isAutoCompactionEnabled() {
+ return hudiTableInfo.getTableType() == HoodieTableType.MERGE_ON_READ
+ &&
hudiTableInfo.getFlinkConfig().get(FlinkOptions.COMPACTION_SCHEDULE_ENABLED);
Review Comment:
This uses Hudi's own `FlinkOptions.COMPACTION_SCHEDULE_ENABLED` whereas
`HudiLakeWriter#shouldRunCompaction` (line 174) uses Fluss-side
`tableInfo.getTableConfig().isDataLakeAutoCompaction()`. The two switches must
agree, otherwise the writer can run compaction execution and produce write
stats that the committer then refuses to commit (or vice versa).
Please unify on a single source of truth. Easiest path: in
`HudiWriteTableInfo.create`, set `COMPACTION_SCHEDULE_ENABLED` from
`tableConfig.isDataLakeAutoCompaction()`; or in this method also check the
Fluss flag.
##########
fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/tiering/HudiLakeCommitter.java:
##########
@@ -207,6 +211,53 @@ public void close() throws Exception {
IOUtils.closeQuietly(hudiTableInfo, "hudi table info");
}
+ @Nullable
+ private String commitCompactionAndSchedule(
+ Map<String, HudiWriteStats> compactionWriteStats, Map<String,
String> commitMetadata) {
+ if (!isAutoCompactionEnabled()) {
+ return null;
+ }
+
+ String latestCommittedInstant = null;
+ try {
+ latestCommittedInstant =
+ compactionService.commitCompaction(compactionWriteStats,
commitMetadata);
+ if (latestCommittedInstant != null) {
+ LOG.info("Committed Hudi compaction instant {}.",
latestCommittedInstant);
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to commit Hudi compaction for table {}. "
+ + "The next tiering round can rollback and retry
pending compactions.",
+ hudiTableInfo.getTablePath(),
+ e);
+ }
+
+ try {
+ compactionService.scheduleCompaction();
+ compactionService.markSelectedCompactionsInflight();
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to schedule Hudi compaction for table {}.",
+ hudiTableInfo.getTablePath(),
+ e);
+ }
+ return latestCommittedInstant;
+ }
+
+ private boolean isAutoCompactionEnabled() {
+ return hudiTableInfo.getTableType() == HoodieTableType.MERGE_ON_READ
+ &&
hudiTableInfo.getFlinkConfig().get(FlinkOptions.COMPACTION_SCHEDULE_ENABLED);
+ }
+
+ static String getLatestCommittedInstant(
+ String dataCommitInstant, @Nullable String compactionInstant) {
+ if (compactionInstant == null ||
dataCommitInstant.compareTo(compactionInstant) >= 0) {
+ return dataCommitInstant;
Review Comment:
Two equal instants both go to `dataCommitInstant` because of `>=`. That is
correct, but please add a unit test for the equal-timestamps edge case so the
choice does not regress (instant timestamps in Hudi have millisecond precision
and collisions, while rare, are possible).
--
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]
