Repository: graphql-java/graphql-java (language: java)

Issue:
Silent thread leak on exception in completeValueForList
This is a very specific bug and thus might not be relevant to most people, but i think it should at least cover the possibility and correctly throw the exception/not leak the thread. I am not  sure i fully understand the bug, but i have a reproduction which i will share.

The bug happens when the `completeValueForList` method throws an exception (for example if getting the iterator throws an exception) and at the same time there is a deeper batch loaded level in another branch of the AST. This leads to the deeper batch loader never to be dispatched, because the parent level is never marked as completed (because of the exception).

**To Reproduce**
The crucial part of the reproduction is the `RuntimeException` in the iterator of the overwritten List implementation. For our use case this exception was a `LazyInitializationException` (Hibernate) thrown inside the iterator - so not an uncommon dependency. The error is obviously something which should be avoided, but since this does not throw any error it is quite hard to pinpoint what and where something went wrong.
```
package graphql

import graphql.execution.instrumentation.dataloader.DataLoaderDispatcherInstrumentation
import graphql.schema.DataFetcher
import graphql.schema.DataFetchingEnvironment
import graphql.schema.StaticDataFetcher
import graphql.schema.idl.RuntimeWiring
import org.apache.commons.lang3.concurrent.BasicThreadFactory
import org.dataloader.BatchLoader
import org.dataloader.DataLoader
import org.dataloader.DataLoaderOptions
import org.dataloader.DataLoaderRegistry
import spock.lang.Specification

import java.util.concurrent.CompletableFuture
import java.util.concurrent.CompletionStage
import java.util.concurrent.SynchronousQueue
import java.util.concurrent.ThreadFactory
import java.util.concurrent.ThreadPoolExecutor
import java.util.concurrent.TimeUnit

import static graphql.schema.idl.TypeRuntimeWiring.newTypeWiring
import static graphql.ExecutionInput.newExecutionInput

class Issue2068 extends Specification {
    def "deadlock attempt"() {
        setup:
        def sdl = """
        type Nation {
            name: String
        }
        
        type Toy {
            name: String
        }
        
        type Owner {
            nation: Nation
        }
        
        type Cat {
            toys: [Toy]
        }
        
        type Dog {
            owner: Owner
        }
        
        type Pets {
            cats: [Cat]
            dogs: [Dog]
        }
        
        type Query {
            pets: Pets
        }
        """

        def cats = [['id': "cat-1"]]
        def dogs = [['id': "dog-1"]]

        ThreadFactory threadFactory = new BasicThreadFactory.Builder()
                .namingPattern("resolver-chain-thread-%d").build()
        def executor = new ThreadPoolExecutor(15, 15, 0L,
                TimeUnit.MILLISECONDS, new SynchronousQueue<>(), threadFactory,
                new ThreadPoolExecutor.CallerRunsPolicy())

        DataFetcher nationsDf = { env -> env.getDataLoader("owner.nation").load(env) }
        DataFetcher ownersDf = { env -> env.getDataLoader("dog.owner").load(env) }

        def wiring = RuntimeWiring.newRuntimeWiring()
                .type(newTypeWiring("Query")
                        .dataFetcher("pets", new StaticDataFetcher(['cats': cats, 'dogs': dogs])))
                .type(newTypeWiring("Cat")
                        .dataFetcher("toys", new StaticDataFetcher(new List<Object>() {
                            @Override
                            int size() {
                                return 1
                            }

                            @Override
                            boolean isEmpty() {
                                return false
                            }

                            @Override
                            boolean contains(Object o) {
                                return false
                            }

                            @Override
                            Iterator iterator() {
                                throw new RuntimeException()
                            }

                            @Override
                            Object[] toArray() {
                                return new Object[0]
                            }

                            @Override
                            Object[] toArray(Object[] a) {
                                return null
                            }

                            @Override
                            boolean add(Object o) {
                                return false
                            }

                            @Override
                            boolean remove(Object o) {
                                return false
                            }

                            @Override
                            boolean containsAll(Collection c) {
                                return false
                            }

                            @Override
                            boolean addAll(Collection c) {
                                return false
                            }

                            @Override
                            boolean addAll(int index, Collection c) {
                                return false
                            }

                            @Override
                            boolean removeAll(Collection c) {
                                return false
                            }

                            @Override
                            boolean retainAll(Collection c) {
                                return false
                            }

                            @Override
                            void clear() {

                            }

                            @Override
                            Object get(int index) {
                                return null
                            }

                            @Override
                            Object set(int index, Object element) {
                                return null
                            }

                            @Override
                            void add(int index, Object element) {

                            }

                            @Override
                            Object remove(int index) {
                                return null
                            }

                            @Override
                            int indexOf(Object o) {
                                return 0
                            }

                            @Override
                            int lastIndexOf(Object o) {
                                return 0
                            }

                            @Override
                            ListIterator listIterator() {
                                return null
                            }

                            @Override
                            ListIterator listIterator(int index) {
                                return null
                            }

                            @Override
                            List subList(int fromIndex, int toIndex) {
                                return null
                            }
                        })))
                .type(newTypeWiring("Dog")
                        .dataFetcher("owner", ownersDf))
                .type(newTypeWiring("Owner")
                        .dataFetcher("nation", nationsDf))
                .build()

        def schema = TestUtil.schema(sdl, wiring)

        when:
        def graphql = GraphQL.newGraphQL(schema)
                .instrumentation(new DataLoaderDispatcherInstrumentation())
                .build()

        then: "execution shouldn't hang"
        DataLoaderRegistry dataLoaderRegistry = mkNewDataLoaderRegistry(executor)

        def result = graphql.executeAsync(newExecutionInput()
                .dataLoaderRegistry(dataLoaderRegistry)
                .query("""
                query LoadPets {
                      pets {
                        cats {
                          toys {
                            name
                          }
                        }
                        dogs {
                          owner {
                            nation {
                              name
                            }
                          }
                        }
                      }
                    }
                    """)
                .build())
        // wait for each future to complete and grab the results
        result.whenComplete({ results, error ->
            if (error) {
                throw error
            }
            results.each { assert it.errors.empty }
        }).join()
    }

    private static DataLoaderRegistry mkNewDataLoaderRegistry(executor) {
        def dataLoaderNations = new DataLoader<Object, Object>(new BatchLoader<DataFetchingEnvironment, List<Object>>() {
            @Override
            CompletionStage<List<List<Object>>> load(List<DataFetchingEnvironment> keys) {
                return CompletableFuture.supplyAsync({
                    def nations = []
                    for (int i = 1; i <= 1; i++) {
                        nations.add(['id': "nation-$i", 'name': "nation-$i"])
                    }
                    return nations
                }, executor) as CompletionStage<List<List<Object>>>
            }
        }, DataLoaderOptions.newOptions().setMaxBatchSize(5))

        def dataLoaderOwners = new DataLoader<Object, Object>(new BatchLoader<DataFetchingEnvironment, List<Object>>() {
            @Override
            CompletionStage<List<List<Object>>> load(List<DataFetchingEnvironment> keys) {
                return CompletableFuture.supplyAsync({
                    def owners = []
                    for (int i = 1; i <= 1; i++) {
                        owners.add(['id': "owner-$i"])
                    }
                    return owners
                }, executor) as CompletionStage<List<List<Object>>>
            }
        }, DataLoaderOptions.newOptions().setMaxBatchSize(5))

        def dataLoaderRegistry = new DataLoaderRegistry()
        dataLoaderRegistry.register("dog.owner", dataLoaderOwners)
        dataLoaderRegistry.register("owner.nation", dataLoaderNations)
        dataLoaderRegistry
    }
}

```
Silent thread leak on exception in completeValueForList
This is a very specific bug and thus might not be relevant to most people, but i think it should at least cover the possibility and correctly throw the exception/not leak the thread. I am not  sure i fully understand the bug, but i have a reproduction which i will share.

The bug happens when the `completeValueForList` method throws an exception (for example if getting the iterator throws an exception) and at the same time there is a deeper batch loaded level in another branch of the AST. This leads to the deeper batch loader never to be dispatched, because the parent level is never marked as completed (because of the exception).

**To Reproduce**
The crucial part of the reproduction is the `RuntimeException` in the iterator of the overwritten List implementation. For our use case this exception was a `LazyInitializationException` (Hibernate) thrown inside the iterator - so not an uncommon dependency. The error is obviously something which should be avoided, but since this does not throw any error it is quite hard to pinpoint what and where something went wrong.
```
package graphql

import graphql.execution.instrumentation.dataloader.DataLoaderDispatcherInstrumentation
import graphql.schema.DataFetcher
import graphql.schema.DataFetchingEnvironment
import graphql.schema.StaticDataFetcher
import graphql.schema.idl.RuntimeWiring
import org.apache.commons.lang3.concurrent.BasicThreadFactory
import org.dataloader.BatchLoader
import org.dataloader.DataLoader
import org.dataloader.DataLoaderOptions
import org.dataloader.DataLoaderRegistry
import spock.lang.Specification

import java.util.concurrent.CompletableFuture
import java.util.concurrent.CompletionStage
import java.util.concurrent.SynchronousQueue
import java.util.concurrent.ThreadFactory
import java.util.concurrent.ThreadPoolExecutor
import java.util.concurrent.TimeUnit

import static graphql.schema.idl.TypeRuntimeWiring.newTypeWiring
import static graphql.ExecutionInput.newExecutionInput

class Issue2068 extends Specification {
    def "deadlock attempt"() {
        setup:
        def sdl = """
        type Nation {
            name: String
        }
        
        type Toy {
            name: String
        }
        
        type Owner {
            nation: Nation
        }
        
        type Cat {
            toys: [Toy]
        }
        
        type Dog {
            owner: Owner
        }
        
        type Pets {
            cats: [Cat]
            dogs: [Dog]
        }
        
        type Query {
            pets: Pets
        }
        """

        def cats = [['id': "cat-1"]]
        def dogs = [['id': "dog-1"]]

        ThreadFactory threadFactory = new BasicThreadFactory.Builder()
                .namingPattern("resolver-chain-thread-%d").build()
        def executor = new ThreadPoolExecutor(15, 15, 0L,
                TimeUnit.MILLISECONDS, new SynchronousQueue<>(), threadFactory,
                new ThreadPoolExecutor.CallerRunsPolicy())

        DataFetcher nationsDf = { env -> env.getDataLoader("owner.nation").load(env) }
        DataFetcher ownersDf = { env -> env.getDataLoader("dog.owner").load(env) }

        def wiring = RuntimeWiring.newRuntimeWiring()
                .type(newTypeWiring("Query")
                        .dataFetcher("pets", new StaticDataFetcher(['cats': cats, 'dogs': dogs])))
                .type(newTypeWiring("Cat")
                        .dataFetcher("toys", new StaticDataFetcher(new List<Object>() {
                            @Override
                            int size() {
                                return 1
                            }

                            @Override
                            boolean isEmpty() {
                                return false
                            }

                            @Override
                            boolean contains(Object o) {
                                return false
                            }

                            @Override
                            Iterator iterator() {
                                throw new RuntimeException()
                            }

                            @Override
                            Object[] toArray() {
                                return new Object[0]
                            }

                            @Override
                            Object[] toArray(Object[] a) {
                                return null
                            }

                            @Override
                            boolean add(Object o) {
                                return false
                            }

                            @Override
                            boolean remove(Object o) {
                                return false
                            }

                            @Override
                            boolean containsAll(Collection c) {
                                return false
                            }

                            @Override
                            boolean addAll(Collection c) {
                                return false
                            }

                            @Override
                            boolean addAll(int index, Collection c) {
                                return false
                            }

                            @Override
                            boolean removeAll(Collection c) {
                                return false
                            }

                            @Override
                            boolean retainAll(Collection c) {
                                return false
                            }

                            @Override
                            void clear() {

                            }

                            @Override
                            Object get(int index) {
                                return null
                            }

                            @Override
                            Object set(int index, Object element) {
                                return null
                            }

                            @Override
                            void add(int index, Object element) {

                            }

                            @Override
                            Object remove(int index) {
                                return null
                            }

                            @Override
                            int indexOf(Object o) {
                                return 0
                            }

                            @Override
                            int lastIndexOf(Object o) {
                                return 0
                            }

                            @Override
                            ListIterator listIterator() {
                                return null
                            }

                            @Override
                            ListIterator listIterator(int index) {
                                return null
                            }

                            @Override
                            List subList(int fromIndex, int toIndex) {
                                return null
                            }
                        })))
                .type(newTypeWiring("Dog")
                        .dataFetcher("owner", ownersDf))
                .type(newTypeWiring("Owner")
                        .dataFetcher("nation", nationsDf))
                .build()

        def schema = TestUtil.schema(sdl, wiring)

        when:
        def graphql = GraphQL.newGraphQL(schema)
                .instrumentation(new DataLoaderDispatcherInstrumentation())
                .build()

        then: "execution shouldn't hang"
        DataLoaderRegistry dataLoaderRegistry = mkNewDataLoaderRegistry(executor)

        def result = graphql.executeAsync(newExecutionInput()
                .dataLoaderRegistry(dataLoaderRegistry)
                .query("""
                query LoadPets {
                      pets {
                        cats {
                          toys {
                            name
                          }
                        }
                        dogs {
                          owner {
                            nation {
                              name
                            }
                          }
                        }
                      }
                    }
                    """)
                .build())
        // wait for each future to complete and grab the results
        result.whenComplete({ results, error ->
            if (error) {
                throw error
            }
            results.each { assert it.errors.empty }
        }).join()
    }

    private static DataLoaderRegistry mkNewDataLoaderRegistry(executor) {
        def dataLoaderNations = new DataLoader<Object, Object>(new BatchLoader<DataFetchingEnvironment, List<Object>>() {
            @Override
            CompletionStage<List<List<Object>>> load(List<DataFetchingEnvironment> keys) {
                return CompletableFuture.supplyAsync({
                    def nations = []
                    for (int i = 1; i <= 1; i++) {
                        nations.add(['id': "nation-$i", 'name': "nation-$i"])
                    }
                    return nations
                }, executor) as CompletionStage<List<List<Object>>>
            }
        }, DataLoaderOptions.newOptions().setMaxBatchSize(5))

        def dataLoaderOwners = new DataLoader<Object, Object>(new BatchLoader<DataFetchingEnvironment, List<Object>>() {
            @Override
            CompletionStage<List<List<Object>>> load(List<DataFetchingEnvironment> keys) {
                return CompletableFuture.supplyAsync({
                    def owners = []
                    for (int i = 1; i <= 1; i++) {
                        owners.add(['id': "owner-$i"])
                    }
                    return owners
                }, executor) as CompletionStage<List<List<Object>>>
            }
        }, DataLoaderOptions.newOptions().setMaxBatchSize(5))

        def dataLoaderRegistry = new DataLoaderRegistry()
        dataLoaderRegistry.register("dog.owner", dataLoaderOwners)
        dataLoaderRegistry.register("owner.nation", dataLoaderNations)
        dataLoaderRegistry
    }
}

```


Discussion / hints:
I had some time to do some more digging and i think i might have found the root cause. Although someone more knowledgeable can maybe have a look and then guide me to a way of fixing this.

The `RuntimeException` thrown inside the list iterator causes one future in the `execute` method of `AsyncExecutionStrategy` to be completed exceptionally and thus reaching the code on [this line](https://github.com/graphql-java/graphql-java/blob/master/src/main/java/graphql/execution/AsyncExecutionStrategy.java#L79). Since an exception here never increases the `happenedOnFieldValueCallsPerLevel` in `FieldLevelTrackingApproach` the deeper levels can never be dispatched because `dispatchIfNeeded` expects that `allOnFieldCallsHappened` . This seems to be the reason for thread leakage, because no exception is thrown and the task is never finished.

(i've trimmed the test case a bit)
I had some time to do some more digging and i think i might have found the root cause. Although someone more knowledgeable can maybe have a look and then guide me to a way of fixing this.

The `RuntimeException` thrown inside the list iterator causes one future in the `execute` method of `AsyncExecutionStrategy` to be completed exceptionally and thus reaching the code on [this line](https://github.com/graphql-java/graphql-java/blob/master/src/main/java/graphql/execution/AsyncExecutionStrategy.java#L79). Since an exception here never increases the `happenedOnFieldValueCallsPerLevel` in `FieldLevelTrackingApproach` the deeper levels can never be dispatched because `dispatchIfNeeded` expects that `allOnFieldCallsHappened` . This seems to be the reason for thread leakage, because no exception is thrown and the task is never finished.

(i've trimmed the test case a bit)

Fix patch:
diff --git a/src/main/java/graphql/execution/AsyncExecutionStrategy.java b/src/main/java/graphql/execution/AsyncExecutionStrategy.java
index 9c70ca06ba..3800657680 100644
--- a/src/main/java/graphql/execution/AsyncExecutionStrategy.java
+++ b/src/main/java/graphql/execution/AsyncExecutionStrategy.java
@@ -76,6 +76,7 @@ public CompletableFuture<ExecutionResult> execute(ExecutionContext executionCont
             // if there are any issues with combining/handling the field results,
             // complete the future at all costs and bubble up any thrown exception so
             // the execution does not hang.
+            executionStrategyCtx.onFieldValuesException();
             overallResult.completeExceptionally(ex);
             return null;
         });
diff --git a/src/main/java/graphql/execution/instrumentation/ExecutionStrategyInstrumentationContext.java b/src/main/java/graphql/execution/instrumentation/ExecutionStrategyInstrumentationContext.java
index ff8dab082d..da11626bc3 100644
--- a/src/main/java/graphql/execution/instrumentation/ExecutionStrategyInstrumentationContext.java
+++ b/src/main/java/graphql/execution/instrumentation/ExecutionStrategyInstrumentationContext.java
@@ -13,4 +13,8 @@ default void onFieldValuesInfo(List<FieldValueInfo> fieldValueInfoList) {
 
     }
 
+    default void onFieldValuesException() {
+
+    }
+
 }
diff --git a/src/main/java/graphql/execution/instrumentation/dataloader/FieldLevelTrackingApproach.java b/src/main/java/graphql/execution/instrumentation/dataloader/FieldLevelTrackingApproach.java
index 3794d67cfb..a6a56a9500 100644
--- a/src/main/java/graphql/execution/instrumentation/dataloader/FieldLevelTrackingApproach.java
+++ b/src/main/java/graphql/execution/instrumentation/dataloader/FieldLevelTrackingApproach.java
@@ -156,6 +156,13 @@ public void onFieldValuesInfo(List<FieldValueInfo> fieldValueInfoList) {
                     dispatch();
                 }
             }
+
+            @Override
+            public void onFieldValuesException() {
+                synchronized (callStack) {
+                    callStack.increaseHappenedOnFieldValueCalls(curLevel);
+                }
+            }
         };
     }
 


Test patch:
diff --git a/src/test/groovy/graphql/Issue2068.groovy b/src/test/groovy/graphql/Issue2068.groovy
new file mode 100644
index 0000000000..e9869119f8
--- /dev/null
+++ b/src/test/groovy/graphql/Issue2068.groovy
@@ -0,0 +1,159 @@
+package graphql
+
+import graphql.execution.instrumentation.dataloader.DataLoaderDispatcherInstrumentation
+import graphql.schema.DataFetcher
+import graphql.schema.DataFetchingEnvironment
+import graphql.schema.StaticDataFetcher
+import graphql.schema.idl.RuntimeWiring
+import org.apache.commons.lang3.concurrent.BasicThreadFactory
+import org.dataloader.BatchLoader
+import org.dataloader.DataLoader
+import org.dataloader.DataLoaderOptions
+import org.dataloader.DataLoaderRegistry
+import spock.lang.Specification
+
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.CompletionStage
+import java.util.concurrent.SynchronousQueue
+import java.util.concurrent.ThreadFactory
+import java.util.concurrent.ThreadPoolExecutor
+import java.util.concurrent.TimeUnit
+
+import static graphql.schema.idl.TypeRuntimeWiring.newTypeWiring
+import static graphql.ExecutionInput.newExecutionInput
+
+class Issue2068 extends Specification {
+    def "shouldn't hang on exception in resolveFieldWithInfo"() {
+        setup:
+        def sdl = """
+        type Nation {
+            name: String
+        }
+        
+        type Toy {
+            name: String
+        }
+        
+        type Owner {
+            nation: Nation
+        }
+        
+        type Cat {
+            toys: [Toy]
+        }
+        
+        type Dog {
+            owner: Owner
+        }
+        
+        type Pets {
+            cats: [Cat]
+            dogs: [Dog]
+        }
+        
+        type Query {
+            pets: Pets
+        }
+        """
+
+        def cats = [['id': "cat-1"]]
+        def dogs = [['id': "dog-1"]]
+
+        ThreadFactory threadFactory = new BasicThreadFactory.Builder()
+                .namingPattern("resolver-chain-thread-%d").build()
+        def executor = new ThreadPoolExecutor(15, 15, 0L,
+                TimeUnit.MILLISECONDS, new SynchronousQueue<>(), threadFactory,
+                new ThreadPoolExecutor.CallerRunsPolicy())
+
+        DataFetcher nationsDf = { env -> env.getDataLoader("owner.nation").load(env) }
+        DataFetcher ownersDf = { env -> env.getDataLoader("dog.owner").load(env) }
+
+        def wiring = RuntimeWiring.newRuntimeWiring()
+                .type(newTypeWiring("Query")
+                        .dataFetcher("pets", new StaticDataFetcher(['cats': cats, 'dogs': dogs])))
+                .type(newTypeWiring("Cat")
+                        .dataFetcher("toys", new StaticDataFetcher(new AbstractList() {
+                            @Override
+                            Object get(int i) {
+                                throw new RuntimeException("Simulated failure");
+                            }
+
+                            @Override
+                            int size() {
+                                return 1
+                            }
+                        })))
+                .type(newTypeWiring("Dog")
+                        .dataFetcher("owner", ownersDf))
+                .type(newTypeWiring("Owner")
+                        .dataFetcher("nation", nationsDf))
+                .build()
+
+        def schema = TestUtil.schema(sdl, wiring)
+
+        when:
+        def graphql = GraphQL.newGraphQL(schema)
+                .instrumentation(new DataLoaderDispatcherInstrumentation())
+                .build()
+        DataLoaderRegistry dataLoaderRegistry = mkNewDataLoaderRegistry(executor)
+
+        graphql.execute(newExecutionInput()
+                .dataLoaderRegistry(dataLoaderRegistry)
+                .query("""
+                query LoadPets {
+                      pets {
+                        cats {
+                          toys {
+                            name
+                          }
+                        }
+                        dogs {
+                          owner {
+                            nation {
+                              name
+                            }
+                          }
+                        }
+                      }
+                    }
+                    """)
+                .build())
+
+        then: "execution shouldn't hang"
+        // wait for each future to complete and grab the results
+        thrown(RuntimeException)
+    }
+
+    private static DataLoaderRegistry mkNewDataLoaderRegistry(executor) {
+        def dataLoaderNations = new DataLoader<Object, Object>(new BatchLoader<DataFetchingEnvironment, List<Object>>() {
+            @Override
+            CompletionStage<List<List<Object>>> load(List<DataFetchingEnvironment> keys) {
+                return CompletableFuture.supplyAsync({
+                    def nations = []
+                    for (int i = 1; i <= 1; i++) {
+                        nations.add(['id': "nation-$i", 'name': "nation-$i"])
+                    }
+                    return nations
+                }, executor) as CompletionStage<List<List<Object>>>
+            }
+        }, DataLoaderOptions.newOptions().setMaxBatchSize(5))
+
+        def dataLoaderOwners = new DataLoader<Object, Object>(new BatchLoader<DataFetchingEnvironment, List<Object>>() {
+            @Override
+            CompletionStage<List<List<Object>>> load(List<DataFetchingEnvironment> keys) {
+                return CompletableFuture.supplyAsync({
+                    def owners = []
+                    for (int i = 1; i <= 1; i++) {
+                        owners.add(['id': "owner-$i"])
+                    }
+                    return owners
+                }, executor) as CompletionStage<List<List<Object>>>
+            }
+        }, DataLoaderOptions.newOptions().setMaxBatchSize(5))
+
+        def dataLoaderRegistry = new DataLoaderRegistry()
+        dataLoaderRegistry.register("dog.owner", dataLoaderOwners)
+        dataLoaderRegistry.register("owner.nation", dataLoaderNations)
+        dataLoaderRegistry
+    }
+}