lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-2368] Catch Throwable in BoundedInMemoryExecutor (#3546)

Browse Source 
Co-authored-by: 喻兆靖 <yuzhaojing@bilibili.com>
This commit is contained in:
yuzhaojing
2021-08-26 20:34:05 +08:00
committed by GitHub
parent cc5256a7d8
commit 73fdcf37df
2 changed files with 4 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
hudi-common/src/main/java/org/apache/hudi/common/util/queue/BoundedInMemoryExecutor.java
View File

@@ -90,7 +90,7 @@ public class BoundedInMemoryExecutor<I, O, E> {
try { try {
preExecute(); preExecute();
producer.produce(queue); producer.produce(queue);
} catch (Exception e) { } catch (Throwable e) {
LOG.error("error producing records", e); LOG.error("error producing records", e);
queue.markAsFailed(e); queue.markAsFailed(e);
throw e; throw e;

6
hudi-common/src/main/java/org/apache/hudi/common/util/queue/BoundedInMemoryQueue.java
View File

@@ -78,10 +78,10 @@ public class BoundedInMemoryQueue<I, O> implements Iterable<O> {
private final long memoryLimit; private final long memoryLimit;
/** /**
* it holds the root cause of the exception in case either queueing records * it holds the root cause of the Throwable in case either queueing records
* (consuming from inputIterator) fails or thread reading records from queue fails. * (consuming from inputIterator) fails or thread reading records from queue fails.
*/ */
private final AtomicReference<Exception> hasFailed = new AtomicReference<>(null); private final AtomicReference<Throwable> hasFailed = new AtomicReference<>(null);
/** Used for indicating that all the records from queue are read successfully. **/ /** Used for indicating that all the records from queue are read successfully. **/
private final AtomicBoolean isReadDone = new AtomicBoolean(false); private final AtomicBoolean isReadDone = new AtomicBoolean(false);
@@ -251,7 +251,7 @@ public class BoundedInMemoryQueue<I, O> implements Iterable<O> {
/** /**
* API to allow producers and consumer to communicate termination due to failure. * API to allow producers and consumer to communicate termination due to failure.
*/ */
public void markAsFailed(Exception e) { public void markAsFailed(Throwable e) {
this.hasFailed.set(e); this.hasFailed.set(e);
// release the permits so that if the queueing thread is waiting for permits then it will // release the permits so that if the queueing thread is waiting for permits then it will
// get it. // get it.