You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ja...@apache.org on 2017/06/20 07:29:35 UTC
[29/56] [abbrv] carbondata git commit: Fixed batch load issue count
and synchronization
Fixed batch load issue count and synchronization
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/72235b3f
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/72235b3f
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/72235b3f
Branch: refs/heads/streaming_ingest
Commit: 72235b3f12f867b78dfd769797f16bd48ed2e062
Parents: 00e0f7e
Author: ravipesala <ra...@gmail.com>
Authored: Thu Jun 15 16:32:09 2017 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Thu Jun 15 17:41:27 2017 +0530
----------------------------------------------------------------------
.../sort/impl/UnsafeBatchParallelReadMergeSorterImpl.java | 9 ++++++++-
1 file changed, 8 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/72235b3f/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/UnsafeBatchParallelReadMergeSorterImpl.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/UnsafeBatchParallelReadMergeSorterImpl.java b/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/UnsafeBatchParallelReadMergeSorterImpl.java
index eebaa67..20a560d 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/UnsafeBatchParallelReadMergeSorterImpl.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/UnsafeBatchParallelReadMergeSorterImpl.java
@@ -198,7 +198,7 @@ public class UnsafeBatchParallelReadMergeSorterImpl extends AbstractMergeSorter
ThreadStatusObserver threadStatusObserver) {
this.sortParameters = sortParameters.getCopy();
this.iteratorCount = new AtomicInteger(numberOfThreads);
- this.mergerQueue = new LinkedBlockingQueue<>();
+ this.mergerQueue = new LinkedBlockingQueue<>(1);
this.threadStatusObserver = threadStatusObserver;
createSortDataRows();
}
@@ -255,6 +255,7 @@ public class UnsafeBatchParallelReadMergeSorterImpl extends AbstractMergeSorter
.getThrowable() instanceof CarbonDataLoadingException) {
finalMerger.setStopProcess(true);
mergerQueue.put(finalMerger);
+ return;
}
processRowToNextStep(sortDataRow, sortParameters);
unsafeIntermediateFileMerger.finish();
@@ -271,6 +272,12 @@ public class UnsafeBatchParallelReadMergeSorterImpl extends AbstractMergeSorter
} catch (CarbonSortKeyAndGroupByException e) {
throw new CarbonDataLoadingException(e);
} catch (InterruptedException e) {
+ // if fails to put in queue because of interrupted exception, we can offer to free the main
+ // thread from waiting.
+ if (finalMerger != null) {
+ finalMerger.setStopProcess(true);
+ mergerQueue.offer(finalMerger);
+ }
throw new CarbonDataLoadingException(e);
}
}