This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new c244b8504f8a feat(flink): add compaction failure metrics for early
detection (#19968)
c244b8504f8a is described below
commit c244b8504f8a9be11b64d4a807fdf628ecf0df8a
Author: Peter Huang <[email protected]>
AuthorDate: Tue Sep 15 21:40:58 2026 -0700
feat(flink): add compaction failure metrics for early detection (#19968)
* feat(flink): add compaction failure metrics for early detection
Adds a compactionErrorCount metric to FlinkCompactionMetrics so
compaction failures - including ones swallowed by NonThrownExecutor
when compaction runs asynchronously - become observable.
- FlinkCompactionMetrics: add a Counter compactionErrorCount, register
it in the metric group, and expose markCompactionFailed() to
increment it.
- DataTableCompactHandler: call markCompactionFailed() in the async
executor's failure callback before emitting the failed commit event,
so every async compaction failure is counted.
- TestDataTableCompactHandler: add test coverage for the failure path.
---
.../hudi/metrics/FlinkCompactionMetrics.java | 15 +++
.../compact/handler/DataTableCompactHandler.java | 5 +-
.../handler/TestDataTableCompactHandler.java | 121 +++++++++++++++++++++
3 files changed, 140 insertions(+), 1 deletion(-)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/metrics/FlinkCompactionMetrics.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/metrics/FlinkCompactionMetrics.java
index 89d3a8c36a90..9616d3baac50 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/metrics/FlinkCompactionMetrics.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/metrics/FlinkCompactionMetrics.java
@@ -27,7 +27,9 @@ import org.apache.hudi.sink.compact.CompactionPlanOperator;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
+import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.SimpleCounter;
import java.text.ParseException;
import java.time.Duration;
@@ -72,6 +74,14 @@ public class FlinkCompactionMetrics extends
FlinkWriteMetrics {
*/
private long compactionStateSignal;
+ /**
+ * Counter for the number of compaction operations that failed, including
the ones swallowed
+ * by the {@link org.apache.hudi.sink.utils.NonThrownExecutor} when
compaction runs asynchronously.
+ *
+ * @see org.apache.hudi.sink.compact.CompactOperator
+ */
+ private final Counter compactionErrorCount = new SimpleCounter();
+
public FlinkCompactionMetrics(MetricGroup metricGroup) {
super(metricGroup, HoodieTimeline.COMPACTION_ACTION);
}
@@ -83,6 +93,11 @@ public class FlinkCompactionMetrics extends
FlinkWriteMetrics {
metricGroup.gauge(getMetricsName(actionType, "compactionDelay"), () ->
compactionDelay);
metricGroup.gauge(getMetricsName(actionType, "compactionCost"), () ->
compactionCost);
metricGroup.gauge(getMetricsName(actionType, "compactionStateSignal"), ()
-> compactionStateSignal);
+ metricGroup.counter(getMetricsName(actionType, "compactionErrorCount"),
compactionErrorCount);
+ }
+
+ public void markCompactionFailed() {
+ compactionErrorCount.inc();
}
public void setFirstPendingCompactionInstant(Option<HoodieInstant>
firstPendingCompactionInstant) {
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/DataTableCompactHandler.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/DataTableCompactHandler.java
index b0619a1586ff..f95b39a0d395 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/DataTableCompactHandler.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/handler/DataTableCompactHandler.java
@@ -107,7 +107,10 @@ public class DataTableCompactHandler implements
CompactHandler {
if (executor != null) {
executor.execute(
() -> doCompaction(event, collector, needReloadMetaClient),
- (errMsg, t) -> collector.collect(createFailedCommitEvent(event)),
+ (errMsg, t) -> {
+ compactionMetrics.markCompactionFailed();
+ collector.collect(createFailedCommitEvent(event));
+ },
"Execute compaction for instant %s from task %d", instantTime,
taskID);
} else {
// executes the compaction task synchronously for batch mode.
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/handler/TestDataTableCompactHandler.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/handler/TestDataTableCompactHandler.java
new file mode 100644
index 000000000000..6053fdb410bf
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/compact/handler/TestDataTableCompactHandler.java
@@ -0,0 +1,121 @@
+/*
+ * 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.sink.compact.handler;
+
+import org.apache.hudi.client.HoodieFlinkWriteClient;
+import org.apache.hudi.common.model.CompactionOperation;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.sink.compact.CompactionCommitEvent;
+import org.apache.hudi.sink.compact.CompactionPlanEvent;
+import org.apache.hudi.sink.utils.NonThrownExecutor;
+import org.apache.hudi.table.HoodieFlinkTable;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.function.ThrowingRunnable;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests that a compaction failure occurring on the async {@link
NonThrownExecutor} path is
+ * reflected in the {@code compactionErrorCount} metric instead of being
swallowed silently.
+ */
+class TestDataTableCompactHandler {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void testAsyncCompactionFailureIncrementsErrorMetric() throws Exception {
+ HoodieFlinkWriteClient<?> writeClient = mock(HoodieFlinkWriteClient.class);
+ HoodieFlinkTable<?> table = mock(HoodieFlinkTable.class);
+ when(writeClient.getHoodieTable()).thenReturn((HoodieFlinkTable) table);
+
+ RuntimeException compactionFailure = new RuntimeException("compaction
boom");
+ DataTableCompactHandler handler = new DataTableCompactHandler(writeClient,
0) {
+ @Override
+ protected void doCompaction(CompactionPlanEvent event,
org.apache.flink.util.Collector<CompactionCommitEvent> collector, boolean
needReloadMetaClient) throws Exception {
+ throw compactionFailure;
+ }
+ };
+
+ MetricGroup metricGroup = mock(MetricGroup.class);
+ ArgumentCaptor<Counter> counterCaptor =
ArgumentCaptor.forClass(Counter.class);
+ handler.registerMetrics(metricGroup);
+ verify(metricGroup).counter(anyString(), counterCaptor.capture());
+ Counter errorCounter = counterCaptor.getValue();
+ assertEquals(0, errorCounter.getCount());
+
+ CompactionOperation operation = mock(CompactionOperation.class);
+ when(operation.getFileId()).thenReturn("file-1");
+ CompactionPlanEvent event = new CompactionPlanEvent("001", operation, 0,
false, false);
+
+ AtomicReference<CompactionCommitEvent> collected = new AtomicReference<>();
+ Collector<CompactionCommitEvent> collector = new
Collector<CompactionCommitEvent>() {
+ @Override
+ public void collect(CompactionCommitEvent record) {
+ collected.set(record);
+ }
+
+ @Override
+ public void close() {
+ // no-op
+ }
+ };
+
+ NonThrownExecutor executor = new SyncFailFastExecutor();
+ handler.compact(executor, event, collector, false);
+
+ assertEquals(1, errorCounter.getCount(), "compaction failure on the async
path must be reflected in the error metric");
+ assertTrue(collected.get().isFailed(), "a failed commit event must still
be emitted downstream");
+ }
+
+ /**
+ * Executes actions synchronously so the async failure hook runs on the
calling test thread.
+ */
+ private static class SyncFailFastExecutor extends NonThrownExecutor {
+ SyncFailFastExecutor() {
+ super(org.slf4j.LoggerFactory.getLogger(SyncFailFastExecutor.class),
null,
+ (errMsg, t) -> {
+ throw new HoodieException(errMsg, t);
+ }, true);
+ }
+
+ @Override
+ public void execute(ThrowingRunnable<Throwable> action, ExceptionHook
hook, String actionName, Object... actionParams) {
+ try {
+ action.run();
+ } catch (Throwable t) {
+ ExceptionUtils.rethrowIfFatalErrorOrOOM(t);
+ if (hook != null) {
+ hook.apply("error", t);
+ }
+ }
+ }
+ }
+}
