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 45cec1a3bc13 fix(common): propagate stored disruptor consumer failures
(#20026)
45cec1a3bc13 is described below
commit 45cec1a3bc139d1e9d6db9bb75efa20451575cea
Author: Burak Kalaycı <[email protected]>
AuthorDate: Tue Oct 13 06:02:57 2026 +0300
fix(common): propagate stored disruptor consumer failures (#20026)
* fix(common): propagate stored disruptor consumer failures
Disruptor handlers catch consume exceptions and store them, so execute()
can return the finish() result while a fatal failure is still stored.
Check that stored failure after finish() so the caller sees it.
Closes #20016
Signed-off-by: Burak KALAYCI <[email protected]>
* test(common): prefer stored consume failure when finish throws
Add a companion case where consume() and finish() throw different
exceptions, and assert the stored consume failure stays the direct cause.
Signed-off-by: Burak KALAYCI <[email protected]>
---------
Signed-off-by: Burak KALAYCI <[email protected]>
---
.../util/queue/BaseHoodieQueueBasedExecutor.java | 9 +-
.../TestDisruptorExecutorConsumerFailure.java | 113 +++++++++++++++++++++
2 files changed, 121 insertions(+), 1 deletion(-)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/util/queue/BaseHoodieQueueBasedExecutor.java
b/hudi-common/src/main/java/org/apache/hudi/common/util/queue/BaseHoodieQueueBasedExecutor.java
index 6faa59342c67..90a8c8f3bd7a 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/util/queue/BaseHoodieQueueBasedExecutor.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/util/queue/BaseHoodieQueueBasedExecutor.java
@@ -201,7 +201,7 @@ public abstract class BaseHoodieQueueBasedExecutor<I, O, E>
implements HoodieExe
// NOTE: To properly support mode when there's no consumer, we have to
fall back
// to producing future as the trigger for us to shut down the queue
- return allOf(Arrays.asList(producingFuture, consumingFuture))
+ E result = allOf(Arrays.asList(producingFuture, consumingFuture))
.whenComplete((ignored, throwable) -> {
// Close the queue to release the resources
queue.close();
@@ -209,6 +209,13 @@ public abstract class BaseHoodieQueueBasedExecutor<I, O,
E> implements HoodieExe
.thenApply(ignored -> consumer.get().finish())
// Block until producing and consuming both finish
.get();
+ // Disruptor handlers store consumer failures and return normally, so
production
+ // and finish() can both succeed while a fatal consume error is still
stored.
+ Throwable consumerFailure = queue.getThrowable();
+ if (consumerFailure != null) {
+ throw new HoodieException(consumerFailure);
+ }
+ return result;
} catch (Exception e) {
if (e instanceof InterruptedException) {
// In case {@code InterruptedException} was thrown, resetting the
interrupted flag
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/util/queue/TestDisruptorExecutorConsumerFailure.java
b/hudi-common/src/test/java/org/apache/hudi/common/util/queue/TestDisruptorExecutorConsumerFailure.java
new file mode 100644
index 000000000000..65c34f620574
--- /dev/null
+++
b/hudi-common/src/test/java/org/apache/hudi/common/util/queue/TestDisruptorExecutorConsumerFailure.java
@@ -0,0 +1,113 @@
+/*
+ * 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.common.util.queue;
+
+import org.apache.hudi.exception.HoodieException;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests that a Disruptor consumer failure reaches {@link
DisruptorExecutor#execute()}
+ * even when production and {@link HoodieConsumer#finish()} complete normally.
+ */
+public class TestDisruptorExecutorConsumerFailure {
+
+ @Test
+ @Timeout(30)
+ public void testExecutePropagatesStoredConsumerFailureAfterFinish() {
+ IllegalStateException consumeFailure = new IllegalStateException("consume
failed");
+ AtomicBoolean finished = new AtomicBoolean(false);
+ HoodieConsumer<String, String> consumer = new HoodieConsumer<String,
String>() {
+ @Override
+ public void consume(String record) {
+ throw consumeFailure;
+ }
+
+ @Override
+ public String finish() {
+ finished.set(true);
+ return "finished";
+ }
+ };
+
+ DisruptorExecutor<String, String, String> executor = new
DisruptorExecutor<>(
+ 8,
+ Collections.singletonList("row").iterator(),
+ consumer,
+ record -> record,
+ WaitStrategyFactory.DEFAULT_STRATEGY,
+ () -> { });
+
+ try {
+ HoodieException thrown = assertThrows(HoodieException.class,
executor::execute);
+ assertSame(consumeFailure, thrown.getCause());
+ assertTrue(finished.get(), "finish() must still release consumer
resources");
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ @Timeout(30)
+ public void testExecutePrefersStoredConsumerFailureWhenFinishThrows() {
+ IllegalStateException consumeFailure = new IllegalStateException("consume
failed");
+ IllegalArgumentException finishFailure = new
IllegalArgumentException("finish failed");
+ // shutdownNow() calls finish() again; only the execute() call should fail.
+ AtomicBoolean finishThrew = new AtomicBoolean(false);
+ HoodieConsumer<String, String> consumer = new HoodieConsumer<String,
String>() {
+ @Override
+ public void consume(String record) {
+ throw consumeFailure;
+ }
+
+ @Override
+ public String finish() {
+ if (finishThrew.compareAndSet(false, true)) {
+ throw finishFailure;
+ }
+ return "closed";
+ }
+ };
+
+ DisruptorExecutor<String, String, String> executor = new
DisruptorExecutor<>(
+ 8,
+ Collections.singletonList("row").iterator(),
+ consumer,
+ record -> record,
+ WaitStrategyFactory.DEFAULT_STRATEGY,
+ () -> { });
+
+ try {
+ HoodieException thrown = assertThrows(HoodieException.class,
executor::execute);
+ assertSame(consumeFailure, thrown.getCause());
+ assertTrue(finishThrew.get(), "finish() must run so the catch path can
prefer the stored failure");
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+}
