You are viewing a plain text version of this content. The canonical link for it is here.

Posted to commits@hudi.apache.org by "kalayciburak (via GitHub)" <gi...@apache.org> on 2026/09/22 14:43:11 UTC

[PR] fix(common): propagate stored disruptor consumer failures [hudi]

kalayciburak opened a new pull request, #20026:
URL: https://github.com/apache/hudi/pull/20026

### Describe the issue this Pull Request addresses

Closes #20016.

`DisruptorMessageQueue` catches exceptions from `HoodieConsumer.consume` and stores them with `markAsFailed`. `BaseHoodieQueueBasedExecutor.execute()` only inspected that stored failure inside its catch block. When production and `finish()` both complete normally, the caller received the finish result and the fatal consume error was dropped.

### Summary and Changelog

After `finish()` returns, `execute()` now throws the stored queue failure if one is present.

- Keep the existing close and `finish()` order so queue and consumer resources are still released.
- Propagate the original consumer exception through the existing `HoodieException` path.
- Add a Disruptor regression test where `consume()` throws and `finish()` returns normally.

### Impact

Callers of `DisruptorExecutor` with `hoodie.write.executor.type=DISRUPTOR` now see a fatal consume failure instead of a successful finish result. No public API, config, or storage-format change. The success path is unchanged when no failure is stored.

### Risk Level

low

The check runs only after both futures complete and only when `queue.getThrowable()` is already set. Verification: `TestDisruptorExecutorConsumerFailure` failed before the change because `execute()` returned normally, then passed (1 test, 0 failures). `hudi-common` checkstyle reported 0 violations.

### Documentation Update

none

### Contributor's checklist

- [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] Adequate tests were added if applicable

--
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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] fix(common): propagate stored disruptor consumer failures [hudi]

Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.

hudi-bot commented on PR #20026:
URL: https://github.com/apache/hudi/pull/20026#issuecomment-5789705969

   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "37607822054d30e90cb1c339eb47fe5e6ea49278",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17238",
       "triggerID" : "37607822054d30e90cb1c339eb47fe5e6ea49278",
       "triggerType" : "PUSH"
     }, {
       "hash" : "08cd0c79216211e0a0c3372af3b7af8b827282c1",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "08cd0c79216211e0a0c3372af3b7af8b827282c1",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 37607822054d30e90cb1c339eb47fe5e6ea49278 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17238) 
   * 08cd0c79216211e0a0c3372af3b7af8b827282c1 UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     @hudi-bot supports the following commands:
   
    - `@hudi-bot run azure` re-run the last Azure build
   </details>


-- 
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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] fix(common): propagate stored disruptor consumer failures [hudi]

Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.

hudi-agent commented on code in PR #20026:
URL: https://github.com/apache/hudi/pull/20026#discussion_r4079415095


##########
hudi-common/src/main/java/org/apache/hudi/common/util/queue/BaseHoodieQueueBasedExecutor.java:
##########
@@ -201,14 +201,21 @@ public E execute() {
 
       // 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();
           })
           .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);

Review Comment:
   🤖 When `finish()` also throws, that exception is discarded entirely (the catch below prefers the stored failure) and never logged — the new test pins that behavior in. Would it be worth attaching it via `addSuppressed` so a failing `finish()` (e.g. a write handle failing to close) still shows up in the stack trace?
   
   <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>



-- 
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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] fix(common): propagate stored disruptor consumer failures [hudi]

Posted by "danny0405 (via GitHub)" <gi...@apache.org>.

danny0405 merged PR #20026:
URL: https://github.com/apache/hudi/pull/20026


-- 
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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] fix(common): propagate stored disruptor consumer failures [hudi]

Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.

hudi-bot commented on PR #20026:
URL: https://github.com/apache/hudi/pull/20026#issuecomment-5790861858

   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "37607822054d30e90cb1c339eb47fe5e6ea49278",
       "status" : "DELETED",
       "url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17238",
       "triggerID" : "37607822054d30e90cb1c339eb47fe5e6ea49278",
       "triggerType" : "PUSH"
     }, {
       "hash" : "08cd0c79216211e0a0c3372af3b7af8b827282c1",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17249",
       "triggerID" : "08cd0c79216211e0a0c3372af3b7af8b827282c1",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 08cd0c79216211e0a0c3372af3b7af8b827282c1 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17249) 
   
   <details>
   <summary>Bot commands</summary>
     @hudi-bot supports the following commands:
   
    - `@hudi-bot run azure` re-run the last Azure build
   </details>


-- 
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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] fix(common): propagate stored disruptor consumer failures [hudi]

Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.

hudi-agent commented on code in PR #20026:
URL: https://github.com/apache/hudi/pull/20026#discussion_r4073006251


##########
hudi-common/src/main/java/org/apache/hudi/common/util/queue/BaseHoodieQueueBasedExecutor.java:
##########
@@ -201,14 +201,21 @@ public E execute() {
 
       // 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();
           })
           .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();

Review Comment:
   🤖 Since `setHandlers` swallows the failure and keeps consuming, every remaining record is still fed to a consumer that has already blown up, and `finish()` then runs over that state. Would it be worth halting the handler (or short-circuiting `consume`) once `throwable` is set, so we don't keep writing after the first fatal error?
   
   <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>



-- 
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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] fix(common): propagate stored disruptor consumer failures [hudi]

Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.

hudi-bot commented on PR #20026:
URL: https://github.com/apache/hudi/pull/20026#issuecomment-5778969710

   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "37607822054d30e90cb1c339eb47fe5e6ea49278",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17238",
       "triggerID" : "37607822054d30e90cb1c339eb47fe5e6ea49278",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 37607822054d30e90cb1c339eb47fe5e6ea49278 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17238) 
   
   <details>
   <summary>Bot commands</summary>
     @hudi-bot supports the following commands:
   
    - `@hudi-bot run azure` re-run the last Azure build
   </details>


-- 
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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] fix(common): propagate stored disruptor consumer failures [hudi]

Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.

hudi-bot commented on PR #20026:
URL: https://github.com/apache/hudi/pull/20026#issuecomment-5778867365

   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "37607822054d30e90cb1c339eb47fe5e6ea49278",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "37607822054d30e90cb1c339eb47fe5e6ea49278",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 37607822054d30e90cb1c339eb47fe5e6ea49278 UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     @hudi-bot supports the following commands:
   
    - `@hudi-bot run azure` re-run the last Azure build
   </details>


-- 
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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] fix(common): propagate stored disruptor consumer failures [hudi]

Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.

hudi-bot commented on PR #20026:
URL: https://github.com/apache/hudi/pull/20026#issuecomment-5780591180

   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "37607822054d30e90cb1c339eb47fe5e6ea49278",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17238",
       "triggerID" : "37607822054d30e90cb1c339eb47fe5e6ea49278",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 37607822054d30e90cb1c339eb47fe5e6ea49278 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17238) 
   
   <details>
   <summary>Bot commands</summary>
     @hudi-bot supports the following commands:
   
    - `@hudi-bot run azure` re-run the last Azure build
   </details>


-- 
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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] fix(common): propagate stored disruptor consumer failures [hudi]

Posted by "danny0405 (via GitHub)" <gi...@apache.org>.

danny0405 commented on code in PR #20026:
URL: https://github.com/apache/hudi/pull/20026#discussion_r4078627398


##########
hudi-common/src/test/java/org/apache/hudi/common/util/queue/TestDisruptorExecutorConsumerFailure.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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());

Review Comment:
   Optional test coverage: could we add a companion case where `consume()` throws this exception and `finish()` throws a different exception, and assert that `consumeFailure` is still the direct cause? This test covers normal finalization, while the companion case would protect the exception-precedence contract in the existing catch path. I checked that scenario locally and the current implementation preserves the original consumer failure, so this is not a blocker.



-- 
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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] fix(common): propagate stored disruptor consumer failures [hudi]

Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.

codecov-commenter commented on PR #20026:
URL: https://github.com/apache/hudi/pull/20026#issuecomment-5788343501

   ## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/20026?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   :x: Patch coverage is `0%` with `5 lines` in your changes missing coverage. Please review.
   :white_check_mark: Project coverage is 80.35%. Comparing base ([`319afce`](https://app.codecov.io/gh/apache/hudi/commit/319afce31cc3677abbb3a5b97a8d3add141ef534?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`3760782`](https://app.codecov.io/gh/apache/hudi/commit/37607822054d30e90cb1c339eb47fe5e6ea49278?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
   :warning: Report is 1 commits behind head on master.
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/hudi/pull/20026?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
   |---|---|---|
   | [...ommon/util/queue/BaseHoodieQueueBasedExecutor.java](https://app.codecov.io/gh/apache/hudi/pull/20026?src=pr&el=tree&filepath=hudi-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fhudi%2Fcommon%2Futil%2Fqueue%2FBaseHoodieQueueBasedExecutor.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL3V0aWwvcXVldWUvQmFzZUhvb2RpZVF1ZXVlQmFzZWRFeGVjdXRvci5qYXZh) | 0.00% | [5 Missing :warning: ](https://app.codecov.io/gh/apache/hudi/pull/20026?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   
   <details><summary>Additional details and impacted files</summary>
   
   
   
   ```diff
   @@             Coverage Diff              @@
   ##             master   #20026      +/-   ##
   ============================================
   - Coverage     80.35%   80.35%   -0.01%     
   - Complexity    34781    34782       +1     
   ============================================
     Files          2545     2545              
     Lines        142646   142650       +4     
     Branches      17327    17328       +1     
   ============================================
   + Hits         114623   114624       +1     
   - Misses        20120    20124       +4     
   + Partials       7903     7902       -1     
   ```
   
   | [Components](https://app.codecov.io/gh/apache/hudi/pull/20026/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [hudi-common](https://app.codecov.io/gh/apache/hudi/pull/20026/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.86% <0.00%> (-0.01%)` | :arrow_down: |
   | [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/20026/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.40% <ø> (ø)` | |
   | [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/20026/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.69% <ø> (ø)` | |
   | [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/20026/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.79% <ø> (ø)` | |
   | [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/20026/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.18% <ø> (ø)` | |
   | [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/20026/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `69.99% <ø> (ø)` | |
   | [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/20026/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.98% <ø> (ø)` | |
   | [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/20026/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.02% <ø> (ø)` | |
   | [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/20026/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.48% <ø> (ø)` | |
   | [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/20026/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.34% <ø> (ø)` | |
   | [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/20026/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.00% <ø> (ø)` | |
   | [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/20026/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `53.20% <ø> (ø)` | |
   
   | [Flag](https://app.codecov.io/gh/apache/hudi/pull/20026/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [hadoop-mr-java-client](https://app.codecov.io/gh/apache/hudi/pull/20026/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `43.86% <0.00%> (+<0.01%)` | :arrow_up: |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/hudi/pull/20026?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [...ommon/util/queue/BaseHoodieQueueBasedExecutor.java](https://app.codecov.io/gh/apache/hudi/pull/20026?src=pr&el=tree&filepath=hudi-common%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fhudi%2Fcommon%2Futil%2Fqueue%2FBaseHoodieQueueBasedExecutor.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1jb21tb24vc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2h1ZGkvY29tbW9uL3V0aWwvcXVldWUvQmFzZUhvb2RpZVF1ZXVlQmFzZWRFeGVjdXRvci5qYXZh) | `86.30% <0.00%> (-6.46%)` | :arrow_down: |
   
   ... and [1 file with indirect coverage changes](https://app.codecov.io/gh/apache/hudi/pull/20026/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
   </details>
   <details><summary> :rocket: New features to boost your workflow: </summary>
   
   - :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
   - :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
   </details>


-- 
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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] fix(common): propagate stored disruptor consumer failures [hudi]

Posted by "kalayciburak (via GitHub)" <gi...@apache.org>.

kalayciburak commented on code in PR #20026:
URL: https://github.com/apache/hudi/pull/20026#discussion_r4079358605


##########
hudi-common/src/test/java/org/apache/hudi/common/util/queue/TestDisruptorExecutorConsumerFailure.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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());

Review Comment:
   added the companion case. finish() throws a different exception, consumeFailure stays the direct cause.



-- 
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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] fix(common): propagate stored disruptor consumer failures [hudi]

Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.

hudi-bot commented on PR #20026:
URL: https://github.com/apache/hudi/pull/20026#issuecomment-5789889968

   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "37607822054d30e90cb1c339eb47fe5e6ea49278",
       "status" : "SUCCESS",
       "url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17238",
       "triggerID" : "37607822054d30e90cb1c339eb47fe5e6ea49278",
       "triggerType" : "PUSH"
     }, {
       "hash" : "08cd0c79216211e0a0c3372af3b7af8b827282c1",
       "status" : "PENDING",
       "url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17249",
       "triggerID" : "08cd0c79216211e0a0c3372af3b7af8b827282c1",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 37607822054d30e90cb1c339eb47fe5e6ea49278 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17238) 
   * 08cd0c79216211e0a0c3372af3b7af8b827282c1 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=17249) 
   
   <details>
   <summary>Bot commands</summary>
     @hudi-bot supports the following commands:
   
    - `@hudi-bot run azure` re-run the last Azure build
   </details>


-- 
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: commits-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org