This is an automated email from the ASF dual-hosted git repository.
akashrn5 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/carbondata.git
The following commit(s) were added to refs/heads/master by this push:
new 1ccfb9b [CARBONDATA-3958] Avoid blocking the loading task when the
output queue poll timeout
1ccfb9b is described below
commit 1ccfb9be3fab8dfc2556541872632471d733518a
Author: QiangCai <qiang...@qq.com>
AuthorDate: Mon Aug 24 16:15:54 2020 +0800
[CARBONDATA-3958] Avoid blocking the loading task when the output queue
poll timeout
Why is this PR needed?
In some cases, the CDC merge tasks are blocked when data loading uses
CarbonTableOutputFormat.getRecordWriter method. PR #3856 change the code to
avoid using
CarbonTableOutputFormat.getRecordWriter, CDC merge will not happen this
issue again.
But this issue maybe happen in other similar scenarios.
Because the poll method of the queue is time out in some cases, so the row
batch in
the queue will not be polled again.
After the queue is full, the queue blocks the writing task puts a new
batch. Even if
the queue is not full, it will also lose the row batch in the queue.
What changes were proposed in this PR?
If the output is not closed, it will poll a row batch in the loops till it
gets a not
null batch.
If the output is closed, it will break the loop.
Does this PR introduce any user interface change?
No
Is any new testcase added?
No
This closes #3897
---
.../iterator/CarbonOutputIteratorWrapper.java | 32 ++++++++++++++++------
1 file changed, 24 insertions(+), 8 deletions(-)
diff --git
a/processing/src/main/java/org/apache/carbondata/processing/loading/iterator/CarbonOutputIteratorWrapper.java
b/processing/src/main/java/org/apache/carbondata/processing/loading/iterator/CarbonOutputIteratorWrapper.java
index 3e29288..dc9aa54 100644
---
a/processing/src/main/java/org/apache/carbondata/processing/loading/iterator/CarbonOutputIteratorWrapper.java
+++
b/processing/src/main/java/org/apache/carbondata/processing/loading/iterator/CarbonOutputIteratorWrapper.java
@@ -63,18 +63,33 @@ public class CarbonOutputIteratorWrapper extends
CarbonIterator<Object[]> {
@Override
public boolean hasNext() {
if (readBatch == null || !readBatch.hasNext()) {
+ // if readBatch don't have next row, set it to null
+ readBatch = null;
try {
- if (!close) {
+ // if the output is not closed, poll a batch from the queue in a loop.
+ // if the queue is always empty, it will wait for the last default
element of the output,
+ // after that, the output will be closed and the loop will be finished.
+ while (!close) {
readBatch = queue.poll(5, TimeUnit.MINUTES);
if (readBatch == null) {
- LOG.warn("This scenario should not happen");
- return false;
+ LOG.warn("try to poll a row batch again.");
+ } else {
+ // if readBatch is not null, break this loop to continue
+ break;
}
- } else {
+ }
+
+ // when the output is closed and readBatch is null, should poll a
batch immediately.
+ // it is a double-checking also of the last poll operation. in some
cases, the output is
+ // closed and the readBatch is null, but the queue is not empty,
contain the last load
+ // batch or the last default batch.
+ if (close && readBatch == null) {
+ LOG.warn("try to poll a row batch one more time.");
readBatch = queue.poll();
- if (readBatch == null) {
- return false;
- }
+ }
+ // checking again whether readBatch is null or not
+ if (readBatch == null) {
+ return false;
}
} catch (InterruptedException e) {
throw new RuntimeException(e);
@@ -112,13 +127,14 @@ public class CarbonOutputIteratorWrapper extends
CarbonIterator<Object[]> {
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
- close = true;
// It is required if the thread waits for take.
if (queue.isEmpty()) {
if (!queue.offer(new RowBatch(0))) {
LOG.warn("The default last element is not added to queue");
}
}
+ // after try to add the default last element, close the output.
+ close = true;
}
private static class RowBatch extends CarbonIterator<Object[]> {
