sohami commented on a change in pull request #1504: DRILL-6792: Find the right
probe side fragment wrapper & fix DrillBuf…
URL: https://github.com/apache/drill/pull/1504#discussion_r236406316
##########
File path:
exec/java-exec/src/main/java/org/apache/drill/exec/work/filter/RuntimeFilterSink.java
##########
@@ -74,32 +70,43 @@
private SendingAccountor sendingAccountor;
+ private AsyncAggregateWorker asyncAggregateWorker;
+
+ private AtomicBoolean running = new AtomicBoolean(true);
+
private static final Logger logger =
LoggerFactory.getLogger(RuntimeFilterSink.class);
public RuntimeFilterSink(DrillbitContext drillbitContext, SendingAccountor
sendingAccountor)
{
this.drillbitContext = drillbitContext;
this.sendingAccountor = sendingAccountor;
- AsyncAggregateWorker asyncAggregateWorker = new AsyncAggregateWorker();
- drillbitContext.getExecutor().submit(asyncAggregateWorker);
}
public void add(RuntimeFilterWritable runtimeFilterWritable)
{
+ if (asyncAggregateWorker == null) {
+ asyncAggregateWorker = new AsyncAggregateWorker();
+ drillbitContext.getExecutor().submit(asyncAggregateWorker);
+ }
runtimeFilterWritable.retainBuffers(1);
int joinMjId =
runtimeFilterWritable.getRuntimeFilterBDef().getMajorFragmentId();
if (joinMjId2Stopwatch.get(joinMjId) == null) {
Stopwatch stopwatch = Stopwatch.createStarted();
joinMjId2Stopwatch.put(joinMjId, stopwatch);
}
- queueLock.lock();
- try {
+ synchronized (rfQueue) {
rfQueue.add(runtimeFilterWritable);
Review comment:
should check `running` status before adding in queue. In cases when running
is false the thread should `close` the received filter
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
