You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by GitBox <gi...@apache.org> on 2020/06/12 07:25:08 UTC
[GitHub] [druid] chenyuzhi459 opened a new pull request #10027: fix query memory leak
chenyuzhi459 opened a new pull request #10027:
URL: https://github.com/apache/druid/pull/10027
### Description
This PR fix a bug where we can not release resources when accumulate someone segments.
### Reason
Because we can not cancel all futures in the `com.google.common.util.concurrent.Futures.CombinedFuture` immediately when exception occurs in someone future by using `CombinedFuture.cancel(boolean mayInterruptIfRunning) ` simply.
Let's see a code example:
```
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
public class GuavaFutureTest {
public static void main(String[] args) {
ExecutorService service = Executors.newFixedThreadPool(3);
ListeningExecutorService exc = MoreExecutors.listeningDecorator(service);
int tasks = 3;
int cancelCount = 10;
AtomicInteger index = new AtomicInteger(0);
Function<Integer, List<ListenableFuture<Object>>> function = (c) -> {
List<ListenableFuture<Object>> futures = new ArrayList<>();
for(int i = 0; i < c; i++){
ListenableFuture<Object> future = exc.submit(new Callable<Object>() {
@Override
public Object call() throws Exception {
AtomicBoolean interupted = new AtomicBoolean(false);
while (true && !interupted.get()){
try {
if(index.get() == cancelCount){
//here we simulate occurs exception in some on future.
throw new RuntimeException("A big bug");
}
//print something to say the task still active.
System.out.println(String.format("Thread[id=%s] running. %s", Thread.currentThread().getId(), System.currentTimeMillis()));
Thread.sleep(1000);
index.getAndIncrement();
} catch (InterruptedException e) {
interupted.set(true);
e.printStackTrace();
}
}
return null;
}
});
futures.add(future);
}
return futures;
};
List<ListenableFuture<Object>> futures = function.apply(tasks);
ListenableFuture future = Futures.allAsList(futures);
try{
future.get();
}catch(Exception e){
System.err.println(e);
// here we try to cancel all tasks. however, we can see all tasks is printing message.
future.cancel(true);
}
}
}
```
Here is the console messages:
```
Thread[id=15] running. 1591945237078
Thread[id=13] running. 1591945237078
Thread[id=14] running. 1591945238078
Thread[id=13] running. 1591945241988
all task should be cancelled.
Thread[id=14] running. 1591945241988
java.util.concurrent.ExecutionException: java.lang.RuntimeException: A big bug
Thread[id=13] running. 1591945242988
Thread[id=14] running. 1591945242988
Thread[id=13] running. 1591945243988
Thread[id=14] running. 1591945243989
```
I had test it from guava[version=16.0.1] to the latest guava[version=29.0]. it's the same result.
Accords to source code in guava[version=16.0.1], when someone future occurs error, it will set the state of combineFuture from `RUNNING` to `COMPLETED` immediately(method stack `CombinedFuture.setOneValue() ->CombinedFuture.setExceptionAndMaybeLog() -> AbstractFuture.setException() -> AbstractFuture.Sync.setException() -> AbstractFuture.Sync.complete()`). Thus when we use `cancel()` method which try to set state from `RUNNING` to ''INTERRUPTED|COMPLETED" through `CAS` operation to stop all tasks, it will failed, because the state is not `RUNNING` anymore.
### Solution
we cancel all compute task manually.
<hr>
This PR has:
- [ ] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] added documentation for new or modified features or behaviors.
- [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/licenses.yaml)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [ ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [ ] been tested in a test Druid cluster.
<!-- Check the items by putting "x" in the brackets for the done things. Not all of these items apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items from the checklist above are strictly necessary, but it would be very helpful if you at least self-review the PR. -->
<hr>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] suneet-s commented on pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
suneet-s commented on pull request #10027:
URL: https://github.com/apache/druid/pull/10027#issuecomment-646148028
@chenyuzhi459 There are more details on the code coverage requirements and how to run the tests locally here - https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md#running-code-coverage-locally
https://travis-ci.org/github/apache/druid/jobs/699729079 - this job is failing because line 104 in
BackgroundCachePopulator is untested. Maybe you can add a new test class `BackgroundCachePopulatorTest` that adds a test that ensures the onFailure method is called.
```
------------------------------------------------------------------------------
org/apache/druid/client/cache/BackgroundCachePopulator.java
------------------------------------------------------------------------------
27 import com.google.common.util.concurrent.ListenableFuture;
28 import com.google.common.util.concurrent.ListeningExecutorService;
29 import com.google.common.util.concurrent.MoreExecutors;
30 F import org.apache.druid.common.guava.GuavaUtils;
104 F | L GuavaUtils.cancelAll(cacheFutures);
------------------------------------------------------------------------------
```
I realize this test coverage was missing before your change, but it would help Druid raise the bar on testing if you are able to add a test for this.
Hope this helps.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] suneet-s commented on pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
suneet-s commented on pull request #10027:
URL: https://github.com/apache/druid/pull/10027#issuecomment-648663220
> According to the job log, it seems the test of `DruidCoordinatorTest.testComputeUnderReplicationCountsPerDataSourcePerTierForSegmentsWithBroadcastRule` doesn't pass.
Actually, the logs indicate that the failing test is `JettyTest.testNumConnectionsMetricHttp` - this is a known flaky test. I've re-triggered the build so hopefully it will succeed in the next run. The logs say `DruidCoordinatorTest.testComputeUnderReplicationCountsPerDataSourcePerTierForSegmentsWithBroadcastRule` failed on the first attempt, but passed on the second attempt, so the test is considered a success.
```
[INFO] Results:
[INFO]
[ERROR] Errors:
[ERROR] org.apache.druid.server.initialization.JettyTest.testNumConnectionsMetricHttp(org.apache.druid.server.initialization.JettyTest)
[ERROR] Run 1: JettyTest.testNumConnectionsMetricHttp:456->waitForJettyServerModuleActiveConnectionsZero:520 Runtime
[ERROR] Run 2: JettyTest.testNumConnectionsMetricHttp:456->waitForJettyServerModuleActiveConnectionsZero:520 Runtime
[ERROR] Run 3: JettyTest.testNumConnectionsMetricHttp:456->waitForJettyServerModuleActiveConnectionsZero:520 Runtime
[ERROR] Run 4: JettyTest.testNumConnectionsMetricHttp:456->waitForJettyServerModuleActiveConnectionsZero:520 Runtime
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 commented on pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 commented on pull request #10027:
URL: https://github.com/apache/druid/pull/10027#issuecomment-650483200
> Thank you for updating the PR . Please fix the checkstyle error.
ok, it's fixed
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r441927349
##########
File path: core/src/main/java/org/apache/druid/common/guava/GuavaUtils.java
##########
@@ -77,4 +79,24 @@ public static Long tryParseLong(@Nullable String string)
}
return arg1;
}
+
+ /**
+ * Cancel futures manually, because sometime we can't cancel all futures in {@link com.google.common.util.concurrent.Futures.CombinedFuture}
+ * automatically. Especially when we call {@link com.google.common.util.concurrent.Futures#allAsList(Iterable)} to create a batch of
+ * future.
+ * @param futures The futures that we want to cancel
+ * @param <T> The result type returned by this Future's {@code get} method
+ */
+ public static <T, F extends Future<T>> void cancelAll(List<F> futures){
+ if(futures == null || futures.isEmpty()){
+ return;
+ }
+ futures.forEach(f -> {
+ try {
+ f.cancel(true);
+ } catch (Throwable t){
+ //do nothing and continue the loop.
Review comment:
Well, I think it's great that we shouldn't chomp throwables. But the method that `AbstractFuture.cancel` probably throws exceptions because it will execute future's listeners which is called by `ExecutionList.executeListener`. And `ExecutionList.executeListener` only catch `RuntimeException`, thus i think we should log exceptions.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 commented on pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 commented on pull request #10027:
URL: https://github.com/apache/druid/pull/10027#issuecomment-648684719
> Actually, the logs indicate that the failing test is `JettyTest.testNumConnectionsMetricHttp` - this is a known flaky test. I've re-triggered the build so hopefully it will succeed in the next run. The logs say `DruidCoordinatorTest.testComputeUnderReplicationCountsPerDataSourcePerTierForSegmentsWithBroadcastRule` failed on the first attempt, but passed on the second attempt, so the test is considered a success.
>
> ```
> [INFO] Results:
> [INFO]
> [ERROR] Errors:
> [ERROR] org.apache.druid.server.initialization.JettyTest.testNumConnectionsMetricHttp(org.apache.druid.server.initialization.JettyTest)
> [ERROR] Run 1: JettyTest.testNumConnectionsMetricHttp:456->waitForJettyServerModuleActiveConnectionsZero:520 Runtime
> [ERROR] Run 2: JettyTest.testNumConnectionsMetricHttp:456->waitForJettyServerModuleActiveConnectionsZero:520 Runtime
> [ERROR] Run 3: JettyTest.testNumConnectionsMetricHttp:456->waitForJettyServerModuleActiveConnectionsZero:520 Runtime
> [ERROR] Run 4: JettyTest.testNumConnectionsMetricHttp:456->waitForJettyServerModuleActiveConnectionsZero:520 Runtime
> ```
Thanks for your help! It pass finally.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r441639550
##########
File path: processing/src/main/java/org/apache/druid/query/ChainedExecutionQueryRunner.java
##########
@@ -141,34 +142,38 @@ public ChainedExecutionQueryRunner(
);
}
)
- )
- );
+ );
- queryWatcher.registerQueryFuture(query, futures);
+ Function<Throwable, Void> cancelFunction = (t) -> {
Review comment:
> This little block is used in a few places, and it would be good to include comments about why it's needed.
>
> Could you please move it to GuavaUtils and also make the following changes:
>
> * Add a comment about why it's needed: it's an alternative to `Futures.allAsList(...).cancel`, that is necessary because `Futures.allAsList` creates a Future that cannot be canceled if one of its constituent futures has already failed. (This comment is the real reason that it's good to have it be its own function.)
> * The function isn't doing anything with the Throwable, so it could just be `void cancelAll(List<Future<T>>)`.
> * Add unit tests to GuavaUtilsTest.
>
> Then we can do stuff like `GuavaUtils.cancelAll(futures)`.
Thanks for your advice, I've revised it.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] clintropolis merged pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
clintropolis merged pull request #10027:
URL: https://github.com/apache/druid/pull/10027
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 edited a comment on pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 edited a comment on pull request #10027:
URL: https://github.com/apache/druid/pull/10027#issuecomment-648612311
> @chenyuzhi459 I skimmed the test jobs. It looks like the processing module tests pass the code coverage now with your new tests.
>
> ```
> Diff coverage statistics:
> ------------------------------------------------------------------------------
> | lines | branches | functions | path
> ------------------------------------------------------------------------------
> | 66% (4/6) | 100% (0/0) | 77% (7/9) | org/apache/druid/query/GroupByMergedQueryRunner.java
> | 87% (7/8) | 100% (0/0) | 83% (5/6) | org/apache/druid/query/ChainedExecutionQueryRunner.java
> | 50% (3/6) | 100% (0/0) | 100% (4/4) | org/apache/druid/query/groupby/epinephelinae/ConcurrentGrouper.java
> | 71% (5/7) | 100% (0/0) | 100% (7/7) | org/apache/druid/query/groupby/epinephelinae/GroupByMergingQueryRunnerV2.java
> ------------------------------------------------------------------------------
> ```
>
> however, it looks like there is a test failure in `ChainedExecutionQueryRunnerTest#testQueryTimeout`
>
> ```
> testQueryTimeout(org.apache.druid.query.ChainedExecutionQueryRunnerTest) Time elapsed: 60.018 s <<< ERROR!
> org.junit.runners.model.TestTimedOutException: test timed out after 60000 milliseconds
> ```
>
> The server module tests are failing on test coverage. I haven't looked closely at how the tests are setup in this PR to validate whether or not the lack of coverage that it's flagging is legitimate
Thanks for your guidance, I have fix it. And now I have only one fail test https://travis-ci.org/github/apache/druid/jobs/701495490.
According to the job log, it seems the test of `DruidCoordinatorTest.testComputeUnderReplicationCountsPerDataSourcePerTierForSegmentsWithBroadcastRule` doesn't pass.
But when I try to run with command:` mvn test -pl server -Pskip-static-checks -Ddruid.console.skip=true -Dmaven.javadoc.skip=true -Dremoteresources.skip=true -Ddruid.generic.useDefaultValueForNull=true` locally, it pass. Could give me more tips, I will try my best to fix it. Thanks!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r446471072
##########
File path: core/src/main/java/org/apache/druid/common/guava/GuavaUtils.java
##########
@@ -23,8 +23,8 @@
import com.google.common.base.Strings;
import com.google.common.primitives.Longs;
import org.apache.druid.java.util.common.logger.Logger;
-
import javax.annotation.Nullable;
+import java.util.ArrayList;
Review comment:
ok, it's done.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] jihoonson commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
jihoonson commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r445720081
##########
File path: processing/src/main/java/org/apache/druid/query/ChainedExecutionQueryRunner.java
##########
@@ -141,33 +144,34 @@ public ChainedExecutionQueryRunner(
);
}
)
- )
- );
+ );
- queryWatcher.registerQueryFuture(query, futures);
+ ListenableFuture<List<Iterable<T>>> future = Futures.allAsList(futures);
+ queryWatcher.registerQueryFuture(query, future);
try {
return new MergeIterable<>(
ordering.nullsFirst(),
QueryContexts.hasTimeout(query) ?
- futures.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
- futures.get()
+ future.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
+ future.get()
).iterator();
}
catch (InterruptedException e) {
log.noStackTrace().warn(e, "Query interrupted, cancelling pending results, query id [%s]", query.getId());
- futures.cancel(true);
+ GuavaUtils.cancelAll(true, ImmutableList.<Future>builder().add(future).addAll(futures).build());
Review comment:
It seems easy to forget canceling `future` and so error-prone. How about modifying `GuavaUtils.cancelAll()` to take `future` as well? So it would be like
```java
public static <F extends Future<?>> void cancelAll(
boolean mayInterruptIfRunning,
@Nullable ListenableFuture<?> combinedFuture,
List<F> futures
)
{
final List<Future> allFuturesToCancel = new ArrayList<>(futures);
allFuturesToCancel.add(combinedFuture);
if (allFuturesToCancel.isEmpty()) {
return;
}
allFuturesToCancel.forEach(f -> {
try {
f.cancel(mayInterruptIfRunning);
}
catch (Throwable t) {
log.warn(t, "Error while cancelling future.");
}
});
}
```
##########
File path: processing/src/main/java/org/apache/druid/query/ChainedExecutionQueryRunner.java
##########
@@ -141,33 +144,34 @@ public ChainedExecutionQueryRunner(
);
}
)
- )
- );
+ );
- queryWatcher.registerQueryFuture(query, futures);
+ ListenableFuture<List<Iterable<T>>> future = Futures.allAsList(futures);
+ queryWatcher.registerQueryFuture(query, future);
try {
return new MergeIterable<>(
ordering.nullsFirst(),
QueryContexts.hasTimeout(query) ?
- futures.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
- futures.get()
+ future.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
+ future.get()
).iterator();
}
catch (InterruptedException e) {
log.noStackTrace().warn(e, "Query interrupted, cancelling pending results, query id [%s]", query.getId());
- futures.cancel(true);
+ GuavaUtils.cancelAll(true, ImmutableList.<Future>builder().add(future).addAll(futures).build());
Review comment:
Or, more structured way to do could be adding a new `CombinedFuture` like this
```java
public static class CombinedFuture<V> implements Future<List<V>>
{
private final List<ListenableFuture<V>> underlyingFutures;
private final ListenableFuture<List<V>> combined;
public CombinedFuture(List<ListenableFuture<V>> futures)
{
this.underlyingFutures = futures;
this.combined = Futures.allAsList(futures);
}
@Override
public boolean cancel(boolean mayInterruptIfRunning)
{
if (combined.isDone() || combined.isCancelled()) {
return false;
} else {
cancelAll(mayInterruptIfRunning, combined, underlyingFutures);
return true;
}
}
@Override
public boolean isCancelled()
{
return combined.isCancelled();
}
@Override
public boolean isDone()
{
return combined.isDone();
}
@Override
public List<V> get() throws InterruptedException, ExecutionException
{
return combined.get();
}
@Override
public List<V> get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
{
return combined.get(timeout, unit);
}
}
```
I'm fine with either way.
##########
File path: processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerFailureTest.java
##########
@@ -281,4 +281,41 @@ public void testInsufficientResourcesOnBroker()
}
}
}
+
+ @Test(timeout = 60_000L)
+ public void testTimeoutExceptionOnQueryable()
+ {
+ expectedException.expect(QueryInterruptedException.class);
+ expectedException.expectCause(CoreMatchers.instanceOf(TimeoutException.class));
+
+ final GroupByQuery query = GroupByQuery
+ .builder()
+ .setDataSource(QueryRunnerTestHelper.DATA_SOURCE)
+ .setQuerySegmentSpec(QueryRunnerTestHelper.FIRST_TO_THIRD)
+ .setDimensions(new DefaultDimensionSpec("quality", "alias"))
+ .setAggregatorSpecs(new LongSumAggregatorFactory("rows", "rows"))
+ .setGranularity(QueryRunnerTestHelper.DAY_GRAN)
+ .overrideContext(ImmutableMap.of(QueryContexts.TIMEOUT_KEY, 1))
+ .build();
+
+ GroupByQueryRunnerFactory factory = makeQueryRunnerFactory(
+ GroupByQueryRunnerTest.DEFAULT_MAPPER,
+ new GroupByQueryConfig()
+ {
+ @Override
+ public String getDefaultStrategy()
+ {
+ return "v2";
+ }
+
+ @Override
+ public boolean isSingleThreaded()
+ {
+ return true;
+ }
+ }
+ );
+ QueryRunner<ResultRow> _runnnner = factory.mergeRunners(Execs.directExecutor(), ImmutableList.of(runner));
Review comment:
nit: the variable name starting with an underscore is not Java convention. How about `mergedRunner`?
##########
File path: processing/src/main/java/org/apache/druid/query/GroupByMergedQueryRunner.java
##########
@@ -187,7 +189,7 @@ private void waitForFutureCompletion(
}
catch (InterruptedException e) {
log.warn(e, "Query interrupted, cancelling pending results, query id [%s]", query.getId());
- future.cancel(true);
+ GuavaUtils.cancelAll(true, futures);
Review comment:
nit: `future` should be canceled on exceptions too. This is nit since this class is used only by groupBy v1 which is deprecated.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r446210819
##########
File path: processing/src/main/java/org/apache/druid/query/GroupByMergedQueryRunner.java
##########
@@ -187,7 +189,7 @@ private void waitForFutureCompletion(
}
catch (InterruptedException e) {
log.warn(e, "Query interrupted, cancelling pending results, query id [%s]", query.getId());
- future.cancel(true);
+ GuavaUtils.cancelAll(true, futures);
Review comment:
Thanks for your tip, i forgot it and now it have been fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r446193150
##########
File path: processing/src/main/java/org/apache/druid/query/ChainedExecutionQueryRunner.java
##########
@@ -141,33 +144,34 @@ public ChainedExecutionQueryRunner(
);
}
)
- )
- );
+ );
- queryWatcher.registerQueryFuture(query, futures);
+ ListenableFuture<List<Iterable<T>>> future = Futures.allAsList(futures);
+ queryWatcher.registerQueryFuture(query, future);
try {
return new MergeIterable<>(
ordering.nullsFirst(),
QueryContexts.hasTimeout(query) ?
- futures.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
- futures.get()
+ future.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
+ future.get()
).iterator();
}
catch (InterruptedException e) {
log.noStackTrace().warn(e, "Query interrupted, cancelling pending results, query id [%s]", query.getId());
- futures.cancel(true);
+ GuavaUtils.cancelAll(true, ImmutableList.<Future>builder().add(future).addAll(futures).build());
Review comment:
> It seems easy to forget canceling `future` and so error-prone. How about modifying `GuavaUtils.cancelAll()` to take `future` as well? So it would be like
>
> ```java
> public static <F extends Future<?>> void cancelAll(
> boolean mayInterruptIfRunning,
> @Nullable ListenableFuture<?> combinedFuture,
> List<F> futures
> )
> {
> final List<Future> allFuturesToCancel = new ArrayList<>(futures);
> allFuturesToCancel.add(combinedFuture);
> if (allFuturesToCancel.isEmpty()) {
> return;
> }
> allFuturesToCancel.forEach(f -> {
> try {
> f.cancel(mayInterruptIfRunning);
> }
> catch (Throwable t) {
> log.warn(t, "Error while cancelling future.");
> }
> });
> }
> ```
Well, thanks for your tips, i'll follow it except one point. It's better to cancel the `combinedFuture` first, because if we cancel the first future in `underlyingFutures` , it will trigger the listener of `com.google.common.util.concurrent.Futures.CombinedFuture` which is added for every future in `underlyingFutures` by `init` method. This listener is actually a method called `setOneValue` which will set combinedFuture's status as `CANCELLED` rather than `INTERRUTED` as we expect when we cancel the first future of underlyingFutures. In addition , the listener of `combinedFuture` will set the status of other future in `underlyingFutures` as the same with itself(`CANCELLED` rather than `INTERRUTED` as we expect). I have test it in the test of `testQueryTimeout` in `ChainedExecutionQueryRunnerTest` use the sequences of [underlyingFutures, combinedFuture].
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] suneet-s edited a comment on pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
suneet-s edited a comment on pull request #10027:
URL: https://github.com/apache/druid/pull/10027#issuecomment-647686948
@chenyuzhi459 I skimmed the test jobs. It looks like the processing module tests pass the code coverage now with your new tests.
```
Diff coverage statistics:
------------------------------------------------------------------------------
| lines | branches | functions | path
------------------------------------------------------------------------------
| 66% (4/6) | 100% (0/0) | 77% (7/9) | org/apache/druid/query/GroupByMergedQueryRunner.java
| 87% (7/8) | 100% (0/0) | 83% (5/6) | org/apache/druid/query/ChainedExecutionQueryRunner.java
| 50% (3/6) | 100% (0/0) | 100% (4/4) | org/apache/druid/query/groupby/epinephelinae/ConcurrentGrouper.java
| 71% (5/7) | 100% (0/0) | 100% (7/7) | org/apache/druid/query/groupby/epinephelinae/GroupByMergingQueryRunnerV2.java
------------------------------------------------------------------------------
```
however, it looks like there is a test failure in `ChainedExecutionQueryRunnerTest#testQueryTimeout`
```
testQueryTimeout(org.apache.druid.query.ChainedExecutionQueryRunnerTest) Time elapsed: 60.018 s <<< ERROR!
org.junit.runners.model.TestTimedOutException: test timed out after 60000 milliseconds
```
The server module tests are failing on test coverage. I haven't looked closely at how the tests are setup in this PR to validate whether or not the lack of coverage that it's flagging is legitimate
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r446193150
##########
File path: processing/src/main/java/org/apache/druid/query/ChainedExecutionQueryRunner.java
##########
@@ -141,33 +144,34 @@ public ChainedExecutionQueryRunner(
);
}
)
- )
- );
+ );
- queryWatcher.registerQueryFuture(query, futures);
+ ListenableFuture<List<Iterable<T>>> future = Futures.allAsList(futures);
+ queryWatcher.registerQueryFuture(query, future);
try {
return new MergeIterable<>(
ordering.nullsFirst(),
QueryContexts.hasTimeout(query) ?
- futures.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
- futures.get()
+ future.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
+ future.get()
).iterator();
}
catch (InterruptedException e) {
log.noStackTrace().warn(e, "Query interrupted, cancelling pending results, query id [%s]", query.getId());
- futures.cancel(true);
+ GuavaUtils.cancelAll(true, ImmutableList.<Future>builder().add(future).addAll(futures).build());
Review comment:
> It seems easy to forget canceling `future` and so error-prone. How about modifying `GuavaUtils.cancelAll()` to take `future` as well? So it would be like
>
> ```java
> public static <F extends Future<?>> void cancelAll(
> boolean mayInterruptIfRunning,
> @Nullable ListenableFuture<?> combinedFuture,
> List<F> futures
> )
> {
> final List<Future> allFuturesToCancel = new ArrayList<>(futures);
> allFuturesToCancel.add(combinedFuture);
> if (allFuturesToCancel.isEmpty()) {
> return;
> }
> allFuturesToCancel.forEach(f -> {
> try {
> f.cancel(mayInterruptIfRunning);
> }
> catch (Throwable t) {
> log.warn(t, "Error while cancelling future.");
> }
> });
> }
> ```
Well, thanks for your tips, i'll follow it except one point. It's better to cancel the `combinedFuture` first, because if we cancel the first future in `underlyingFutures` , it will trigger the listener of `com.google.common.util.concurrent.Futures.CombinedFuture` which is added for every future in `underlyingFutures` by `init` method. This listener is actually a method called `setOneValue` which will set combinedFuture's status as `CANCELLED` rather than `INTERRUTED` as we expect when we cancel the first future of underlyingFutures cause of the `CancellationException`.
In addition , the listener of `combinedFuture` will set the status of other future in `underlyingFutures` as the same with itself(`CANCELLED` rather than `INTERRUTED` as we expect). I have test it in the test of `testQueryTimeout` in `ChainedExecutionQueryRunnerTest` use the sequences of [underlyingFutures, combinedFuture], it failed cause the second future was not `INTERRUTED` but `CANCELLED`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] jihoonson commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
jihoonson commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r446303931
##########
File path: core/src/main/java/org/apache/druid/common/guava/GuavaUtils.java
##########
@@ -23,8 +23,8 @@
import com.google.common.base.Strings;
import com.google.common.primitives.Longs;
import org.apache.druid.java.util.common.logger.Logger;
-
import javax.annotation.Nullable;
+import java.util.ArrayList;
Review comment:
```[ERROR] /home/travis/build/apache/druid/core/src/main/java/org/apache/druid/common/guava/GuavaUtils.java:26: 'javax.annotation.Nullable' should be separated from previous imports. [ImportOrder]```
The checkstyle wants an empty line between the Lines 26 and 27.
##########
File path: processing/src/main/java/org/apache/druid/query/ChainedExecutionQueryRunner.java
##########
@@ -141,33 +144,34 @@ public ChainedExecutionQueryRunner(
);
}
)
- )
- );
+ );
- queryWatcher.registerQueryFuture(query, futures);
+ ListenableFuture<List<Iterable<T>>> future = Futures.allAsList(futures);
+ queryWatcher.registerQueryFuture(query, future);
try {
return new MergeIterable<>(
ordering.nullsFirst(),
QueryContexts.hasTimeout(query) ?
- futures.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
- futures.get()
+ future.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
+ future.get()
).iterator();
}
catch (InterruptedException e) {
log.noStackTrace().warn(e, "Query interrupted, cancelling pending results, query id [%s]", query.getId());
- futures.cancel(true);
+ GuavaUtils.cancelAll(true, ImmutableList.<Future>builder().add(future).addAll(futures).build());
Review comment:
Ah makes sense 👍. Would you please add a comment on [this line](https://github.com/apache/druid/pull/10027/files/7fc9d5036af8003f1a576ccab128b57571ad03fd..a51d7e16f5f15af33fa239419aaa82bffb99f8f5#diff-298d22d65226e0f32251f009130a0032R102) about why we cancel it first? It would help other people to understand your intention. Maybe the comment can say "Canceling combinedFuture first so that it can complete with `INTERRUPTED` as its final state. See ChainedExecutionQueryRunnerTest.testQueryTimeout()".
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] gianm commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
gianm commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r440572205
##########
File path: processing/src/main/java/org/apache/druid/query/ChainedExecutionQueryRunner.java
##########
@@ -141,34 +142,38 @@ public ChainedExecutionQueryRunner(
);
}
)
- )
- );
+ );
- queryWatcher.registerQueryFuture(query, futures);
+ Function<Throwable, Void> cancelFunction = (t) -> {
Review comment:
This little block is used in a few places, and it would be good to include comments about why it's needed.
Could you please move it to GuavaUtils and also make the following changes:
- Add a comment about why it's needed: it's an alternative to `Futures.allAsList(...).cancel`, that is necessary because `Futures.allAsList` creates a Future that cannot be canceled if one of its constituent futures has already failed. This is the real reason that it's good to have it be its own function.
- The function isn't doing anything with the Throwable, so it could just be `void cancelAll(List<Future<T>>)`.
- Add unit tests to GuavaUtilsTest.
Then we can do stuff like `GuavaUtils.cancelAll(futures)`.
##########
File path: processing/src/main/java/org/apache/druid/query/GroupByMergedQueryRunner.java
##########
@@ -173,11 +173,16 @@ public T apply(Row input)
private void waitForFutureCompletion(
GroupByQuery query,
- ListenableFuture<?> future,
+ List<ListenableFuture<Void>> futures,
IncrementalIndex<?> closeOnFailure
)
{
+ Function<Throwable, Void> cancelFunction = (t) -> {
Review comment:
Similar comments as in ChainedExecutionQueryRunner about GuavaUtils.
##########
File path: processing/src/main/java/org/apache/druid/query/ChainedExecutionQueryRunner.java
##########
@@ -141,34 +142,38 @@ public ChainedExecutionQueryRunner(
);
}
)
- )
- );
+ );
- queryWatcher.registerQueryFuture(query, futures);
+ Function<Throwable, Void> cancelFunction = (t) -> {
+ futures.forEach(f -> f.cancel(true));
Review comment:
Can `f.cancel(true)` throw an exception? I'm not sure, but if so I guess we should catch it and keep canceling more futures.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 commented on pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 commented on pull request #10027:
URL: https://github.com/apache/druid/pull/10027#issuecomment-648612311
> @chenyuzhi459 I skimmed the test jobs. It looks like the processing module tests pass the code coverage now with your new tests.
>
> ```
> Diff coverage statistics:
> ------------------------------------------------------------------------------
> | lines | branches | functions | path
> ------------------------------------------------------------------------------
> | 66% (4/6) | 100% (0/0) | 77% (7/9) | org/apache/druid/query/GroupByMergedQueryRunner.java
> | 87% (7/8) | 100% (0/0) | 83% (5/6) | org/apache/druid/query/ChainedExecutionQueryRunner.java
> | 50% (3/6) | 100% (0/0) | 100% (4/4) | org/apache/druid/query/groupby/epinephelinae/ConcurrentGrouper.java
> | 71% (5/7) | 100% (0/0) | 100% (7/7) | org/apache/druid/query/groupby/epinephelinae/GroupByMergingQueryRunnerV2.java
> ------------------------------------------------------------------------------
> ```
>
> however, it looks like there is a test failure in `ChainedExecutionQueryRunnerTest#testQueryTimeout`
>
> ```
> testQueryTimeout(org.apache.druid.query.ChainedExecutionQueryRunnerTest) Time elapsed: 60.018 s <<< ERROR!
> org.junit.runners.model.TestTimedOutException: test timed out after 60000 milliseconds
> ```
>
> The server module tests are failing on test coverage. I haven't looked closely at how the tests are setup in this PR to validate whether or not the lack of coverage that it's flagging is legitimate
Thanks for your guidance. I have fix it. And now I have only one fail test https://travis-ci.org/github/apache/druid/jobs/701495490.
According to the job log, it seems the test of `DruidCoordinatorTest.testComputeUnderReplicationCountsPerDataSourcePerTierForSegmentsWithBroadcastRule` doesn't pass. But when I try to run with command:` mvn test -pl server -Pskip-static-checks -Ddruid.console.skip=true -Dmaven.javadoc.skip=true -Dremoteresources.skip=true -Ddruid.generic.useDefaultValueForNull=true` locally, it pass. Could give me more tips, I will try my best to fix it. Thanks!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 edited a comment on pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 edited a comment on pull request #10027:
URL: https://github.com/apache/druid/pull/10027#issuecomment-645737656
> Could you please check out the things flagged by CI too: https://travis-ci.org/github/apache/druid/builds/697603463
> Could you please check out the things flagged by CI too: https://travis-ci.org/github/apache/druid/builds/697603463
I am unfamiliar with coverage test. Take https://travis-ci.org/github/apache/druid/jobs/699358945 as an example, I have searched some docs to learn ,but still hard to pass coverage test. Could you give me a little favor
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r446193150
##########
File path: processing/src/main/java/org/apache/druid/query/ChainedExecutionQueryRunner.java
##########
@@ -141,33 +144,34 @@ public ChainedExecutionQueryRunner(
);
}
)
- )
- );
+ );
- queryWatcher.registerQueryFuture(query, futures);
+ ListenableFuture<List<Iterable<T>>> future = Futures.allAsList(futures);
+ queryWatcher.registerQueryFuture(query, future);
try {
return new MergeIterable<>(
ordering.nullsFirst(),
QueryContexts.hasTimeout(query) ?
- futures.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
- futures.get()
+ future.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
+ future.get()
).iterator();
}
catch (InterruptedException e) {
log.noStackTrace().warn(e, "Query interrupted, cancelling pending results, query id [%s]", query.getId());
- futures.cancel(true);
+ GuavaUtils.cancelAll(true, ImmutableList.<Future>builder().add(future).addAll(futures).build());
Review comment:
> It seems easy to forget canceling `future` and so error-prone. How about modifying `GuavaUtils.cancelAll()` to take `future` as well? So it would be like
>
> ```java
> public static <F extends Future<?>> void cancelAll(
> boolean mayInterruptIfRunning,
> @Nullable ListenableFuture<?> combinedFuture,
> List<F> futures
> )
> {
> final List<Future> allFuturesToCancel = new ArrayList<>(futures);
> allFuturesToCancel.add(combinedFuture);
> if (allFuturesToCancel.isEmpty()) {
> return;
> }
> allFuturesToCancel.forEach(f -> {
> try {
> f.cancel(mayInterruptIfRunning);
> }
> catch (Throwable t) {
> log.warn(t, "Error while cancelling future.");
> }
> });
> }
> ```
Well, thanks for your tips, i'll follow it except one point. It's better to cancel the `combinedFuture` first, because if we cancel the first future in `underlyingFutures` , it will trigger the listener of `com.google.common.util.concurrent.Futures.CombinedFuture` which is added for every future in `underlyingFutures` by `init` method. This listener is actually a method called `setOneValue` which will set combinedFuture's status as `CANCELLED` rather than `INTERRUTED` as we expect. In addition , the listener of `combinedFuture` will set the status of other future in `underlyingFutures` as the same with itself(`CANCELLED` rather than `INTERRUTED` as we expect). I have test it in the test of `testQueryTimeout` in `ChainedExecutionQueryRunnerTest`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 commented on pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 commented on pull request #10027:
URL: https://github.com/apache/druid/pull/10027#issuecomment-647626491
> @chenyuzhi459 There are more details on the code coverage requirements and how to run the tests locally here - https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md#running-code-coverage-locally
>
> https://travis-ci.org/github/apache/druid/jobs/699729079 - this job is failing because line 104 in
> BackgroundCachePopulator is untested. Maybe you can add a new test class `BackgroundCachePopulatorTest` that adds a test that ensures the onFailure method is called.
>
> ```
> ------------------------------------------------------------------------------
> org/apache/druid/client/cache/BackgroundCachePopulator.java
> ------------------------------------------------------------------------------
> 27 import com.google.common.util.concurrent.ListenableFuture;
> 28 import com.google.common.util.concurrent.ListeningExecutorService;
> 29 import com.google.common.util.concurrent.MoreExecutors;
> 30 F import org.apache.druid.common.guava.GuavaUtils;
> 104 F | L GuavaUtils.cancelAll(cacheFutures);
> ------------------------------------------------------------------------------
> ```
>
> I realize this test coverage was missing before your change, but it would help Druid raise the bar on testing if you are able to add a test for this.
>
> Hope this helps.
Thanks for your tips， it helps me a lot! I have add a unit test for `BackgroundCachePopulator` named ``
> @chenyuzhi459 There are more details on the code coverage requirements and how to run the tests locally here - https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md#running-code-coverage-locally
>
> https://travis-ci.org/github/apache/druid/jobs/699729079 - this job is failing because line 104 in
> BackgroundCachePopulator is untested. Maybe you can add a new test class `BackgroundCachePopulatorTest` that adds a test that ensures the onFailure method is called.
>
> ```
> ------------------------------------------------------------------------------
> org/apache/druid/client/cache/BackgroundCachePopulator.java
> ------------------------------------------------------------------------------
> 27 import com.google.common.util.concurrent.ListenableFuture;
> 28 import com.google.common.util.concurrent.ListeningExecutorService;
> 29 import com.google.common.util.concurrent.MoreExecutors;
> 30 F import org.apache.druid.common.guava.GuavaUtils;
> 104 F | L GuavaUtils.cancelAll(cacheFutures);
> ------------------------------------------------------------------------------
> ```
>
> I realize this test coverage was missing before your change, but it would help Druid raise the bar on testing if you are able to add a test for this.
>
> Hope this helps.
Thanks for your tips, it helps me a lot. I have added a unit test named `BackgroundCachePopulatorTest` for `BackgroundCachePopulator` , however, it still fail to pass travis ci test. In https://travis-ci.org/github/apache/druid/jobs/700279328, the line coverage for `BackgroundCachePopulator` is still `0%`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] gianm commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
gianm commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r441779534
##########
File path: core/src/main/java/org/apache/druid/common/guava/GuavaUtils.java
##########
@@ -77,4 +79,24 @@ public static Long tryParseLong(@Nullable String string)
}
return arg1;
}
+
+ /**
+ * Cancel futures manually, because sometime we can't cancel all futures in {@link com.google.common.util.concurrent.Futures.CombinedFuture}
+ * automatically. Especially when we call {@link com.google.common.util.concurrent.Futures#allAsList(Iterable)} to create a batch of
+ * future.
+ * @param futures The futures that we want to cancel
+ * @param <T> The result type returned by this Future's {@code get} method
+ */
+ public static <T, F extends Future<T>> void cancelAll(List<F> futures){
+ if(futures == null || futures.isEmpty()){
+ return;
+ }
+ futures.forEach(f -> {
+ try {
+ f.cancel(true);
+ } catch (Throwable t){
+ //do nothing and continue the loop.
Review comment:
We shouldn't chomp Throwables, because they're generally bad things that should interrupt execution. It'd be better to either assume no exceptions, i.e.:
```java
futures.forEach(f -> f.cancel(true));
```
Or surface Errors, but suppress and log exceptions:
```java
futures.forEach(f -> {
try {
f.cancel(true);
} catch (Exception e) {
log.warn(e, "Error while canceling future.");
}
});
```
IMO, the first one is best if we can assure ourselves that `ListenableFuture.cancel` isn't going to throw exceptions. Looking at `AbstractFuture`, it seems like it won't, so the first option looks safe to me.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] suneet-s commented on pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
suneet-s commented on pull request #10027:
URL: https://github.com/apache/druid/pull/10027#issuecomment-647686948
@chenyuzhi459 I skimmed the tests. It looks like the processing module tests pass the code coverage now with your new tests.
```
Diff coverage statistics:
------------------------------------------------------------------------------
| lines | branches | functions | path
------------------------------------------------------------------------------
| 66% (4/6) | 100% (0/0) | 77% (7/9) | org/apache/druid/query/GroupByMergedQueryRunner.java
| 87% (7/8) | 100% (0/0) | 83% (5/6) | org/apache/druid/query/ChainedExecutionQueryRunner.java
| 50% (3/6) | 100% (0/0) | 100% (4/4) | org/apache/druid/query/groupby/epinephelinae/ConcurrentGrouper.java
| 71% (5/7) | 100% (0/0) | 100% (7/7) | org/apache/druid/query/groupby/epinephelinae/GroupByMergingQueryRunnerV2.java
------------------------------------------------------------------------------
```
however, it looks like there is a test failure in `ChainedExecutionQueryRunnerTest#testQueryTimeout`
```
testQueryTimeout(org.apache.druid.query.ChainedExecutionQueryRunnerTest) Time elapsed: 60.018 s <<< ERROR!
org.junit.runners.model.TestTimedOutException: test timed out after 60000 milliseconds
```
The processing tests are failing on test coverage. I haven't looked closely at how the tests are setup in this PR to validate whether or not the lack of coverage that it's flagging is legitimate
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] gianm commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
gianm commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r440572205
##########
File path: processing/src/main/java/org/apache/druid/query/ChainedExecutionQueryRunner.java
##########
@@ -141,34 +142,38 @@ public ChainedExecutionQueryRunner(
);
}
)
- )
- );
+ );
- queryWatcher.registerQueryFuture(query, futures);
+ Function<Throwable, Void> cancelFunction = (t) -> {
Review comment:
This little block is used in a few places, and it would be good to include comments about why it's needed.
Could you please move it to GuavaUtils and also make the following changes:
- Add a comment about why it's needed: it's an alternative to `Futures.allAsList(...).cancel`, that is necessary because `Futures.allAsList` creates a Future that cannot be canceled if one of its constituent futures has already failed. (This comment is the real reason that it's good to have it be its own function.)
- The function isn't doing anything with the Throwable, so it could just be `void cancelAll(List<Future<T>>)`.
- Add unit tests to GuavaUtilsTest.
Then we can do stuff like `GuavaUtils.cancelAll(futures)`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] suneet-s edited a comment on pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
suneet-s edited a comment on pull request #10027:
URL: https://github.com/apache/druid/pull/10027#issuecomment-647686948
@chenyuzhi459 I skimmed the tests. It looks like the processing module tests pass the code coverage now with your new tests.
```
Diff coverage statistics:
------------------------------------------------------------------------------
| lines | branches | functions | path
------------------------------------------------------------------------------
| 66% (4/6) | 100% (0/0) | 77% (7/9) | org/apache/druid/query/GroupByMergedQueryRunner.java
| 87% (7/8) | 100% (0/0) | 83% (5/6) | org/apache/druid/query/ChainedExecutionQueryRunner.java
| 50% (3/6) | 100% (0/0) | 100% (4/4) | org/apache/druid/query/groupby/epinephelinae/ConcurrentGrouper.java
| 71% (5/7) | 100% (0/0) | 100% (7/7) | org/apache/druid/query/groupby/epinephelinae/GroupByMergingQueryRunnerV2.java
------------------------------------------------------------------------------
```
however, it looks like there is a test failure in `ChainedExecutionQueryRunnerTest#testQueryTimeout`
```
testQueryTimeout(org.apache.druid.query.ChainedExecutionQueryRunnerTest) Time elapsed: 60.018 s <<< ERROR!
org.junit.runners.model.TestTimedOutException: test timed out after 60000 milliseconds
```
The server module tests are failing on test coverage. I haven't looked closely at how the tests are setup in this PR to validate whether or not the lack of coverage that it's flagging is legitimate
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 commented on pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 commented on pull request #10027:
URL: https://github.com/apache/druid/pull/10027#issuecomment-645737656
> Could you please check out the things flagged by CI too: https://travis-ci.org/github/apache/druid/builds/697603463
> Could you please check out the things flagged by CI too: https://travis-ci.org/github/apache/druid/builds/697603463
I am unfamiliar with coverage test. Take `https://travis-ci.org/github/apache/druid/jobs/699358945` as an example, I have searched some docs to learn ,but still hard to pass coverage test. Could you give me a little favor
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r446193150
##########
File path: processing/src/main/java/org/apache/druid/query/ChainedExecutionQueryRunner.java
##########
@@ -141,33 +144,34 @@ public ChainedExecutionQueryRunner(
);
}
)
- )
- );
+ );
- queryWatcher.registerQueryFuture(query, futures);
+ ListenableFuture<List<Iterable<T>>> future = Futures.allAsList(futures);
+ queryWatcher.registerQueryFuture(query, future);
try {
return new MergeIterable<>(
ordering.nullsFirst(),
QueryContexts.hasTimeout(query) ?
- futures.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
- futures.get()
+ future.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
+ future.get()
).iterator();
}
catch (InterruptedException e) {
log.noStackTrace().warn(e, "Query interrupted, cancelling pending results, query id [%s]", query.getId());
- futures.cancel(true);
+ GuavaUtils.cancelAll(true, ImmutableList.<Future>builder().add(future).addAll(futures).build());
Review comment:
> It seems easy to forget canceling `future` and so error-prone. How about modifying `GuavaUtils.cancelAll()` to take `future` as well? So it would be like
>
> ```java
> public static <F extends Future<?>> void cancelAll(
> boolean mayInterruptIfRunning,
> @Nullable ListenableFuture<?> combinedFuture,
> List<F> futures
> )
> {
> final List<Future> allFuturesToCancel = new ArrayList<>(futures);
> allFuturesToCancel.add(combinedFuture);
> if (allFuturesToCancel.isEmpty()) {
> return;
> }
> allFuturesToCancel.forEach(f -> {
> try {
> f.cancel(mayInterruptIfRunning);
> }
> catch (Throwable t) {
> log.warn(t, "Error while cancelling future.");
> }
> });
> }
> ```
Well, thanks for your tips, i'll follow it except one point. It's better to cancel the `combinedFuture` first, because if we cancel the first future in `underlyingFutures` , it will trigger the listener of `com.google.common.util.concurrent.Futures.CombinedFuture` which is added for every future in `underlyingFutures` by `init` method. This listener is actually a method called `setOneValue` which will set combinedFuture's status as `CANCELLED` rather than `INTERRUTED` as we expect when we cancel the first future of underlyingFutures.
In addition , the listener of `combinedFuture` will set the status of other future in `underlyingFutures` as the same with itself(`CANCELLED` rather than `INTERRUTED` as we expect). I have test it in the test of `testQueryTimeout` in `ChainedExecutionQueryRunnerTest` use the sequences of [underlyingFutures, combinedFuture], it failed cause the second future was not `INTERRUTED` but `CANCELLED`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r446211225
##########
File path: processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerFailureTest.java
##########
@@ -281,4 +281,41 @@ public void testInsufficientResourcesOnBroker()
}
}
}
+
+ @Test(timeout = 60_000L)
+ public void testTimeoutExceptionOnQueryable()
+ {
+ expectedException.expect(QueryInterruptedException.class);
+ expectedException.expectCause(CoreMatchers.instanceOf(TimeoutException.class));
+
+ final GroupByQuery query = GroupByQuery
+ .builder()
+ .setDataSource(QueryRunnerTestHelper.DATA_SOURCE)
+ .setQuerySegmentSpec(QueryRunnerTestHelper.FIRST_TO_THIRD)
+ .setDimensions(new DefaultDimensionSpec("quality", "alias"))
+ .setAggregatorSpecs(new LongSumAggregatorFactory("rows", "rows"))
+ .setGranularity(QueryRunnerTestHelper.DAY_GRAN)
+ .overrideContext(ImmutableMap.of(QueryContexts.TIMEOUT_KEY, 1))
+ .build();
+
+ GroupByQueryRunnerFactory factory = makeQueryRunnerFactory(
+ GroupByQueryRunnerTest.DEFAULT_MAPPER,
+ new GroupByQueryConfig()
+ {
+ @Override
+ public String getDefaultStrategy()
+ {
+ return "v2";
+ }
+
+ @Override
+ public boolean isSingleThreaded()
+ {
+ return true;
+ }
+ }
+ );
+ QueryRunner<ResultRow> _runnnner = factory.mergeRunners(Execs.directExecutor(), ImmutableList.of(runner));
Review comment:
It have been fixed
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] jihoonson commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
jihoonson commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r441191277
##########
File path: processing/src/main/java/org/apache/druid/query/ChainedExecutionQueryRunner.java
##########
@@ -141,34 +142,38 @@ public ChainedExecutionQueryRunner(
);
}
)
- )
- );
+ );
- queryWatcher.registerQueryFuture(query, futures);
+ Function<Throwable, Void> cancelFunction = (t) -> {
Review comment:
I've been thinking about this, and it seems good to use `CompletableFuture` rather than having another home-grown util method for `ListenableFuture`. But it needs more work like how to support the query priority with `CompletableFuture`, it would be fine to add a util method for now.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] gianm commented on pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
gianm commented on pull request #10027:
URL: https://github.com/apache/druid/pull/10027#issuecomment-644531269
Could you please check out the things flagged by CI too: https://travis-ci.org/github/apache/druid/builds/697603463
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] chenyuzhi459 commented on a change in pull request #10027: fix query memory leak
Posted by GitBox <gi...@apache.org>.
chenyuzhi459 commented on a change in pull request #10027:
URL: https://github.com/apache/druid/pull/10027#discussion_r446193150
##########
File path: processing/src/main/java/org/apache/druid/query/ChainedExecutionQueryRunner.java
##########
@@ -141,33 +144,34 @@ public ChainedExecutionQueryRunner(
);
}
)
- )
- );
+ );
- queryWatcher.registerQueryFuture(query, futures);
+ ListenableFuture<List<Iterable<T>>> future = Futures.allAsList(futures);
+ queryWatcher.registerQueryFuture(query, future);
try {
return new MergeIterable<>(
ordering.nullsFirst(),
QueryContexts.hasTimeout(query) ?
- futures.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
- futures.get()
+ future.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
+ future.get()
).iterator();
}
catch (InterruptedException e) {
log.noStackTrace().warn(e, "Query interrupted, cancelling pending results, query id [%s]", query.getId());
- futures.cancel(true);
+ GuavaUtils.cancelAll(true, ImmutableList.<Future>builder().add(future).addAll(futures).build());
Review comment:
> It seems easy to forget canceling `future` and so error-prone. How about modifying `GuavaUtils.cancelAll()` to take `future` as well? So it would be like
>
> ```java
> public static <F extends Future<?>> void cancelAll(
> boolean mayInterruptIfRunning,
> @Nullable ListenableFuture<?> combinedFuture,
> List<F> futures
> )
> {
> final List<Future> allFuturesToCancel = new ArrayList<>(futures);
> allFuturesToCancel.add(combinedFuture);
> if (allFuturesToCancel.isEmpty()) {
> return;
> }
> allFuturesToCancel.forEach(f -> {
> try {
> f.cancel(mayInterruptIfRunning);
> }
> catch (Throwable t) {
> log.warn(t, "Error while cancelling future.");
> }
> });
> }
> ```
Well, thanks for your tips, i'll follow it except one point. It's better to cancel the `combinedFuture` first, because if we cancel the first future in `underlyingFutures` , it will trigger the listener of `com.google.common.util.concurrent.Futures.CombinedFuture` which is added for every future in `underlyingFutures` by `init` method. This listener is actually a method called `setOneValue` which will set combinedFuture's status as `CANCELLED` rather than `INTERRUTED` as we expect when we cancel the first future of underlyingFutures.
In addition , the listener of `combinedFuture` will set the status of other future in `underlyingFutures` as the same with itself(`CANCELLED` rather than `INTERRUTED` as we expect). I have test it in the test of `testQueryTimeout` in `ChainedExecutionQueryRunnerTest` use the sequences of [underlyingFutures, combinedFuture].
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org