Merge branch '1.0.x'

spring-graphql-docs/src/docs/asciidoc/index.adoc

@@ -258,6 +258,7 @@ the following:
 - Detect https://www.graphql-java.com/documentation/instrumentation[Instrumentation] beans for
   {spring-boot-ref-docs}/actuator.html#actuator.metrics.supported.spring-graphql[GraphQL metrics].
 - Detect `DataFetcherExceptionResolver` beans for <<execution-exceptions, exception resolution>>.
+- Detect `SubscriptionExceptionResolver` beans for <<execution-exceptions-subsctiption, subscription exception resolution>>.
 
 For further customizations, you can declare your own `GraphQlSourceBuilderCustomizer` beans;
 for example, for configuring your own `ExecutionIdProvider`:
@@ -565,6 +566,22 @@ Unresolved exception are logged at ERROR level along with the `executionId` to c
 to the error sent to the client. Resolved exceptions are logged at DEBUG level.
 
 
+[[execution-exceptions-subsctiption]]
+==== Subscription Exceptions
+
+The `Publisher` for a subscription request may complete with an error signal in which case
+the underlying transport (e.g. WebSocket) sends a final "error" type message with a list
+of GraphQL errors.
+
+`DataFetcherExceptionResolver` cannot resolve errors from a subscription `Publisher`,
+since the data `DataFetcher` only creates the `Publisher` initially. After that, the
+transport subscribes to the `Publisher` that may then complete with an error.
+
+An application can register a `SubscriptionExceptionResolver` in order to resolve
+exceptions from a subscription `Publisher` in order to resolve those to GraphQL errors
+to send to the client.
+
+
 
 [[execution-batching]]
 === Batch Loading

spring-graphql/src/main/java/org/springframework/graphql/execution/AbstractGraphQlSourceBuilder.java

@@ -43,6 +43,8 @@ abstract class AbstractGraphQlSourceBuilder<B extends GraphQlSource.Builder<B>>
 
 	private final List<DataFetcherExceptionResolver> exceptionResolvers = new ArrayList<>();
 
+	private final List<SubscriptionExceptionResolver> subscriptionExceptionResolvers = new ArrayList<>();
+
 	private final List<GraphQLTypeVisitor> typeVisitors = new ArrayList<>();
 
 	private final List<Instrumentation> instrumentations = new ArrayList<>();
@@ -57,6 +59,12 @@ abstract class AbstractGraphQlSourceBuilder<B extends GraphQlSource.Builder<B>>
 		return self();
 	}
 
+	@Override
+	public B subscriptionExceptionResolvers(List<SubscriptionExceptionResolver> resolvers) {
+		this.subscriptionExceptionResolvers.addAll(resolvers);
+		return self();
+	}
+
 	@Override
 	public B typeVisitors(List<GraphQLTypeVisitor> typeVisitors) {
 		this.typeVisitors.addAll(typeVisitors);
@@ -105,8 +113,9 @@ abstract class AbstractGraphQlSourceBuilder<B extends GraphQlSource.Builder<B>>
 	protected abstract GraphQLSchema initGraphQlSchema();
 
 	private GraphQLSchema applyTypeVisitors(GraphQLSchema schema) {
+		GraphQLTypeVisitor visitor = ContextDataFetcherDecorator.createVisitor(this.subscriptionExceptionResolvers);
 		List<GraphQLTypeVisitor> visitors = new ArrayList<>(this.typeVisitors);
-		visitors.add(ContextDataFetcherDecorator.TYPE_VISITOR);
+		visitors.add(visitor);
 
 		GraphQLCodeRegistry.Builder codeRegistry = GraphQLCodeRegistry.newCodeRegistry(schema.getCodeRegistry());
 		Map<Class<?>, Object> vars = Collections.singletonMap(GraphQLCodeRegistry.Builder.class, codeRegistry);

spring-graphql/src/main/java/org/springframework/graphql/execution/CompositeSubscriptionExceptionResolver.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2002-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.graphql.execution;
+
+import java.util.Collections;
+import java.util.List;
+
+import graphql.GraphQLError;
+import graphql.GraphqlErrorBuilder;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import org.springframework.util.Assert;
+
+/**
+ * Implementation of {@link SubscriptionExceptionResolver} that is given a list
+ * of other {@link SubscriptionExceptionResolver}'s to invoke in turn until one
+ * returns a list of {@link GraphQLError}'s.
+ *
+ * <p>If the exception remains unresolved, it is mapped to a default error of
+ * type {@link ErrorType#INTERNAL_ERROR} with a generic message.
+ *
+ * @author Mykyta Ivchenko
+ * @author Rossen Stoyanchev
+ * @since 1.0.1
+ */
+class CompositeSubscriptionExceptionResolver implements SubscriptionExceptionResolver {
+
+    private static final Log logger = LogFactory.getLog(CompositeSubscriptionExceptionResolver.class);
+
+    private final List<SubscriptionExceptionResolver> resolvers;
+
+
+    CompositeSubscriptionExceptionResolver(List<SubscriptionExceptionResolver> resolvers) {
+        Assert.notNull(resolvers, "'resolvers' is required");
+        this.resolvers = resolvers;
+    }
+
+
+    @Override
+    public Mono<List<GraphQLError>> resolveException(Throwable exception) {
+        return Flux.fromIterable(this.resolvers)
+                .flatMap(resolver -> resolver.resolveException(exception))
+                .next()
+                .onErrorResume(error -> Mono.just(handleResolverException(error, exception)))
+                .defaultIfEmpty(createDefaultError());
+    }
+
+    private List<GraphQLError> handleResolverException(
+            Throwable resolverException, Throwable originalException) {
+
+        if (logger.isWarnEnabled()) {
+            logger.warn("Failure while resolving " + originalException.getClass().getName(), resolverException);
+        }
+        return createDefaultError();
+    }
+
+    private List<GraphQLError> createDefaultError() {
+        return Collections.singletonList(GraphqlErrorBuilder.newError()
+                .message("Subscription error")
+                .errorType(ErrorType.INTERNAL_ERROR)
+                .build());
+    }
+
+}

spring-graphql/src/main/java/org/springframework/graphql/execution/ContextDataFetcherDecorator.java

@@ -16,6 +16,8 @@
 
 package org.springframework.graphql.execution;
 
+import java.util.List;
+
 import graphql.ExecutionInput;
 import graphql.schema.DataFetcher;
 import graphql.schema.DataFetchingEnvironment;
@@ -41,6 +43,7 @@ import org.springframework.util.Assert;
  * <li>Support {@link Flux} return value as a shortcut to {@link Flux#collectList()}.
  * <li>Re-establish Reactor Context passed via {@link ExecutionInput}.
  * <li>Re-establish ThreadLocal context passed via {@link ExecutionInput}.
+ * <li>Resolve exceptions from a GraphQL subscription {@link Publisher}.
  * </ul>
  *
  * @author Rossen Stoyanchev
@@ -51,10 +54,17 @@ final class ContextDataFetcherDecorator implements DataFetcher<Object> {
 
 	private final boolean subscription;
 
-	private ContextDataFetcherDecorator(DataFetcher<?> delegate, boolean subscription) {
+	private final SubscriptionExceptionResolver subscriptionExceptionResolver;
+
+	private ContextDataFetcherDecorator(
+			DataFetcher<?> delegate, boolean subscription,
+			SubscriptionExceptionResolver subscriptionExceptionResolver) {
+
 		Assert.notNull(delegate, "'delegate' DataFetcher is required");
+		Assert.notNull(subscriptionExceptionResolver, "'subscriptionExceptionResolver' is required");
 		this.delegate = delegate;
 		this.subscription = subscription;
+		this.subscriptionExceptionResolver = subscriptionExceptionResolver;
 	}
 
 	@Override
@@ -66,7 +76,11 @@ final class ContextDataFetcherDecorator implements DataFetcher<Object> {
 		ContextView contextView = ReactorContextManager.getReactorContext(environment.getGraphQlContext());
 
 		if (this.subscription) {
-			return (!contextView.isEmpty() ? Flux.from((Publisher<?>) value).contextWrite(contextView) : value);
+			Assert.state(value instanceof Publisher, "Expected Publisher for a subscription");
+			Flux<?> flux = Flux.from((Publisher<?>) value).onErrorResume(exception ->
+					this.subscriptionExceptionResolver.resolveException(exception)
+							.flatMap(errors -> Mono.error(new SubscriptionPublisherException(errors, exception))));
+			return (!contextView.isEmpty() ? flux.contextWrite(contextView) : flux);
 		}
 
 		if (value instanceof Flux) {
@@ -84,29 +98,34 @@ final class ContextDataFetcherDecorator implements DataFetcher<Object> {
 		return value;
 	}
 
+
 	/**
-	 * {@link GraphQLTypeVisitor} that wraps non-GraphQL data fetchers and adapts them if
-	 * they return {@link Flux} or {@link Mono}.
+	 * Static factory method to create {@link GraphQLTypeVisitor} that wraps
+	 * data fetchers with the {@link ContextDataFetcherDecorator}.
 	 */
-	static GraphQLTypeVisitor TYPE_VISITOR = new GraphQLTypeVisitorStub() {
+	static GraphQLTypeVisitor createVisitor(List<SubscriptionExceptionResolver> resolvers) {
 
-		@Override
-		public TraversalControl visitGraphQLFieldDefinition(GraphQLFieldDefinition fieldDefinition,
-				TraverserContext<GraphQLSchemaElement> context) {
+		SubscriptionExceptionResolver compositeResolver = new CompositeSubscriptionExceptionResolver(resolvers);
 
-			GraphQLCodeRegistry.Builder codeRegistry = context.getVarFromParents(GraphQLCodeRegistry.Builder.class);
-			GraphQLFieldsContainer parent = (GraphQLFieldsContainer) context.getParentNode();
-			DataFetcher<?> dataFetcher = codeRegistry.getDataFetcher(parent, fieldDefinition);
+		return new GraphQLTypeVisitorStub() {
+			@Override
+			public TraversalControl visitGraphQLFieldDefinition(GraphQLFieldDefinition fieldDefinition,
+																TraverserContext<GraphQLSchemaElement> context) {
 
-			if (dataFetcher.getClass().getPackage().getName().startsWith("graphql.")) {
+				GraphQLCodeRegistry.Builder codeRegistry = context.getVarFromParents(GraphQLCodeRegistry.Builder.class);
+				GraphQLFieldsContainer parent = (GraphQLFieldsContainer) context.getParentNode();
+				DataFetcher<?> dataFetcher = codeRegistry.getDataFetcher(parent, fieldDefinition);
+
+				if (dataFetcher.getClass().getPackage().getName().startsWith("graphql.")) {
+					return TraversalControl.CONTINUE;
+				}
+
+				boolean handlesSubscription = parent.getName().equals("Subscription");
+				dataFetcher = new ContextDataFetcherDecorator(dataFetcher, handlesSubscription, compositeResolver);
+				codeRegistry.dataFetcher(parent, fieldDefinition, dataFetcher);
 				return TraversalControl.CONTINUE;
 			}
-
-			boolean handlesSubscription = parent.getName().equals("Subscription");
-			dataFetcher = new ContextDataFetcherDecorator(dataFetcher, handlesSubscription);
-			codeRegistry.dataFetcher(parent, fieldDefinition, dataFetcher);
-			return TraversalControl.CONTINUE;
-		}
-	};
+		};
+	}
 
 }

spring-graphql/src/main/java/org/springframework/graphql/execution/DataFetcherExceptionResolver.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
 package org.springframework.graphql.execution;
 
 import java.util.List;
+import java.util.function.BiFunction;
 
 import graphql.GraphQLError;
 import graphql.schema.DataFetchingEnvironment;
@@ -55,9 +56,31 @@ public interface DataFetcherExceptionResolver {
 	 * @return a {@code Mono} with errors to add to the GraphQL response;
 	 * if the {@code Mono} completes with an empty List, the exception is resolved
 	 * without any errors added to the response; if the {@code Mono} completes
-	 * empty, without emitting a List, the exception remains unresolved and gives
-	 * other resolvers a chance.
+	 * empty, without emitting a List, the exception remains unresolved and that
+	 * allows other resolvers to resolve it.
 	 */
 	Mono<List<GraphQLError>> resolveException(Throwable exception, DataFetchingEnvironment environment);
 
+
+	/**
+	 * Factory method to create a {@link DataFetcherExceptionResolver} to resolve
+	 * an exception to a single GraphQL error. Effectively, a shortcut
+	 * for creating {@link DataFetcherExceptionResolverAdapter} and overriding
+	 * its {@code resolveToSingleError} method.
+	 * @param resolver the resolver function to use
+	 * @return the created instance
+	 * @since 1.0.1
+	 */
+	static DataFetcherExceptionResolverAdapter forSingleError(
+			BiFunction<Throwable, DataFetchingEnvironment, GraphQLError> resolver) {
+
+		return new DataFetcherExceptionResolverAdapter() {
+
+			@Override
+			protected GraphQLError resolveToSingleError(Throwable ex, DataFetchingEnvironment env) {
+				return resolver.apply(ex, env);
+			}
+		};
+	}
+
 }

spring-graphql/src/main/java/org/springframework/graphql/execution/DataFetcherExceptionResolverAdapter.java

@@ -34,14 +34,16 @@ import org.springframework.lang.Nullable;
  * <li>{@link #resolveToMultipleErrors}
  * </ul>
  *
- * <p>Use {@link #from(BiFunction)} to create an instance or extend this class
- * and override one of its resolve methods.
+ * <p>Applications may also use
+ * {@link DataFetcherExceptionResolver#forSingleError(BiFunction)} as a shortcut
+ * for {@link #resolveToSingleError(Throwable, DataFetchingEnvironment)}.
  *
  * <p>Implementations can also express interest in ThreadLocal context
  * propagation, from the underlying transport thread, via
  * {@link #setThreadLocalContextAware(boolean)}.
  *
  * @author Rossen Stoyanchev
+ * @since 1.0.0
  */
 public abstract class DataFetcherExceptionResolverAdapter implements DataFetcherExceptionResolver {
 
@@ -57,7 +59,7 @@ public abstract class DataFetcherExceptionResolverAdapter implements DataFetcher
 
 
 	/**
-	 * Sub-classes can set this to indicate that ThreadLocal context from the
+	 * Subclasses can set this to indicate that ThreadLocal context from the
 	 * transport handler (e.g. HTTP handler) should be restored when resolving
 	 * exceptions.
 	 * <p><strong>Note:</strong> This property is applicable only if transports
@@ -129,7 +131,9 @@ public abstract class DataFetcherExceptionResolverAdapter implements DataFetcher
 	 * resolves exceptions with the given {@code BiFunction}.
 	 * @param resolver the resolver function to use
 	 * @return the created instance
+	 * @deprecated as of 1.0.1, please use {@link DataFetcherExceptionResolver#forSingleError(BiFunction)}
 	 */
+	@Deprecated
 	public static DataFetcherExceptionResolverAdapter from(
 			BiFunction<Throwable, DataFetchingEnvironment, GraphQLError> resolver) {
 

spring-graphql/src/main/java/org/springframework/graphql/execution/ExceptionResolversExceptionHandler.java

@@ -41,6 +41,7 @@ import org.springframework.util.Assert;
  * in a sequence until one returns a list of {@link GraphQLError}'s.
  *
  * @author Rossen Stoyanchev
+ * @since 1.0.0
  */
 class ExceptionResolversExceptionHandler implements DataFetcherExceptionHandler {
 

spring-graphql/src/main/java/org/springframework/graphql/execution/GraphQlSource.java

@@ -16,7 +16,6 @@
 
 package org.springframework.graphql.execution;
 
-import java.io.InputStream;
 import java.util.List;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
@@ -84,13 +83,26 @@ public interface GraphQlSource {
 	interface Builder<B extends Builder<B>> {
 
 		/**
-		 * Add {@link DataFetcherExceptionResolver}s for resolving exceptions
-		 * from {@link graphql.schema.DataFetcher}s.
+		 * Add {@link DataFetcherExceptionResolver}'s that are invoked when a
+		 * {@link graphql.schema.DataFetcher} raises an exception. Resolvers
+		 * are invoked in sequence until one emits a list.
 		 * @param resolvers the resolvers to add
 		 * @return the current builder
 		 */
 		B exceptionResolvers(List<DataFetcherExceptionResolver> resolvers);
 
+		/**
+		 * Add {@link SubscriptionExceptionResolver}s that are invoked when a
+		 * GraphQL subscription {@link org.reactivestreams.Publisher} ends with
+		 * error, and given a chance to resolve the exception to one or more
+		 * GraphQL errors to be sent to the client. Resolvers are invoked in
+		 * sequence until one emits a list.
+		 * @param resolvers the subscription exception resolver
+		 * @return the current builder
+		 * @since 1.0.1
+		 */
+		B subscriptionExceptionResolvers(List<SubscriptionExceptionResolver> resolvers);
+
 		/**
 		 * Add {@link GraphQLTypeVisitor}s to visit all element of the created
 		 * {@link graphql.schema.GraphQLSchema}.
@@ -136,7 +148,7 @@ public interface GraphQlSource {
 
 		/**
 		 * Add schema definition resources, typically {@literal ".graphqls"} files, to be
-		 * {@link graphql.schema.idl.SchemaParser#parse(InputStream) parsed} and
+		 * {@link graphql.schema.idl.SchemaParser#parse(java.io.InputStream) parsed} and
 		 * {@link TypeDefinitionRegistry#merge(TypeDefinitionRegistry) merged}.
 		 * @param resources resources with GraphQL schema definitions
 		 * @return the current builder

spring-graphql/src/main/java/org/springframework/graphql/execution/SubscriptionExceptionResolver.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2002-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.graphql.execution;
+
+import java.util.List;
+import java.util.function.Function;
+
+import graphql.GraphQLError;
+import reactor.core.publisher.Mono;
+
+/**
+ * Contract for a component that is invoked when a GraphQL subscription
+ * {@link org.reactivestreams.Publisher} ends with an error.
+ *
+ * <p>Resolver implementations can extend the convenience base class
+ * {@link SubscriptionExceptionResolverAdapter} and override one of its methods
+ * {@link SubscriptionExceptionResolverAdapter#resolveToSingleError resolveToSingleError} or
+ * {@link SubscriptionExceptionResolverAdapter#resolveToMultipleErrors resolveToMultipleErrors}
+ * that resolve the exception synchronously.
+ *
+ * <p>Resolved errors are wrapped in a {@link SubscriptionPublisherException}
+ * and propagated further to the underlying transport which access the errors
+ * and prepare a final error message to send to the client.
+ *
+ * @author Mykyta Ivchenko
+ * @author Rossen Stoyanchev
+ * @since 1.0.1
+ * @see SubscriptionExceptionResolverAdapter
+ */
+@FunctionalInterface
+public interface SubscriptionExceptionResolver {
+
+    /**
+     * Resolve the given exception to a list of {@link GraphQLError}'s to be
+     * sent in an error message to the client.
+     * @param exception the exception from the Publisher
+     * @return a {@code Mono} with the GraphQL errors to send to the client;
+     * if the {@code Mono} completes with an empty List, the exception is resolved
+     * without any errors to send; if the {@code Mono} completes empty, without
+     * emitting a List, the exception remains unresolved, and that allows other
+     * resolvers to resolve it.
+     */
+    Mono<List<GraphQLError>> resolveException(Throwable exception);
+
+
+    /**
+     * Factory method to create a {@link SubscriptionExceptionResolver} to
+     * resolve an exception to a single GraphQL error. Effectively, a shortcut
+     * for creating {@link SubscriptionExceptionResolverAdapter} and overriding
+     * its {@code resolveToSingleError} method.
+     * @param resolver the resolver function to map the exception
+     * @return the created instance
+     */
+    static SubscriptionExceptionResolverAdapter forSingleError(Function<Throwable, GraphQLError> resolver) {
+        return new SubscriptionExceptionResolverAdapter() {
+
+            @Override
+            protected GraphQLError resolveToSingleError(Throwable ex) {
+                return resolver.apply(ex);
+            }
+        };
+    }
+
+}

spring-graphql/src/main/java/org/springframework/graphql/execution/SubscriptionExceptionResolverAdapter.java

@@ -0,0 +1,114 @@
+/*
+ * Copyright 2002-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.graphql.execution;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
+
+import graphql.GraphQLError;
+import reactor.core.publisher.Mono;
+
+import org.springframework.lang.Nullable;
+
+/**
+ * Adapter for {@link SubscriptionExceptionResolver} that pre-implements the
+ * asynchronous contract and exposes the following synchronous protected methods:
+ * <ul>
+ * <li>{@link #resolveToSingleError}
+ * <li>{@link #resolveToMultipleErrors}
+ * </ul>
+ *
+ * <p>Applications may also use
+ * {@link SubscriptionExceptionResolver#forSingleError(Function)} as a shortcut
+ * for {@link #resolveToSingleError(Throwable)}.
+ *
+ * @author Mykyta Ivchenko
+ * @author Rossen Stoyanchev
+ * @since 1.0.1
+ * @see SubscriptionExceptionResolver
+ */
+public abstract class SubscriptionExceptionResolverAdapter implements SubscriptionExceptionResolver {
+
+    private boolean threadLocalContextAware;
+
+
+    /**
+     * Subclasses can set this to indicate that ThreadLocal context from the
+     * transport handler (e.g. HTTP handler) should be restored when resolving
+     * exceptions.
+     * <p><strong>Note:</strong> This property is applicable only if transports
+     * use ThreadLocal's' (e.g. Spring MVC) and if a {@link ThreadLocalAccessor}
+     * is registered to extract ThreadLocal values of interest. There is no
+     * impact from setting this property otherwise.
+     * <p>By default this is set to "false" in which case there is no attempt
+     * to propagate ThreadLocal context.
+     * @param threadLocalContextAware whether this resolver needs access to
+     * ThreadLocal context or not.
+     */
+    public void setThreadLocalContextAware(boolean threadLocalContextAware) {
+        this.threadLocalContextAware = threadLocalContextAware;
+    }
+
+    /**
+     * Whether ThreadLocal context needs to be restored for this resolver.
+     */
+    public boolean isThreadLocalContextAware() {
+        return this.threadLocalContextAware;
+    }
+
+
+    @Override
+    public final Mono<List<GraphQLError>> resolveException(Throwable exception) {
+        if (!this.threadLocalContextAware) {
+            return Mono.justOrEmpty(resolveToMultipleErrors(exception));
+        }
+        return Mono.deferContextual(contextView -> {
+            List<GraphQLError> errors;
+            try {
+                ReactorContextManager.restoreThreadLocalValues(contextView);
+                errors = resolveToMultipleErrors(exception);
+            }
+            finally {
+                ReactorContextManager.resetThreadLocalValues(contextView);
+            }
+            return Mono.justOrEmpty(errors);
+        });
+    }
+
+    /**
+     * Override this method to resolve the Exception to multiple GraphQL errors.
+     * @param exception the exception to resolve
+     * @return the resolved errors or {@code null} if unresolved
+     */
+    @Nullable
+    protected List<GraphQLError> resolveToMultipleErrors(Throwable exception) {
+        GraphQLError error = resolveToSingleError(exception);
+        return (error != null ? Collections.singletonList(error) : null);
+    }
+
+    /**
+     * Override this method to resolve the Exception to a single GraphQL error.
+     * @param exception the exception to resolve
+     * @return the resolved error or {@code null} if unresolved
+     */
+    @Nullable
+    protected GraphQLError resolveToSingleError(Throwable exception) {
+        return null;
+    }
+
+}

spring-graphql/src/main/java/org/springframework/graphql/execution/SubscriptionPublisherException.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2002-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.graphql.execution;
+
+import java.util.List;
+
+import graphql.GraphQLError;
+
+import org.springframework.core.NestedRuntimeException;
+
+/**
+ * An exception raised after a GraphQL subscription
+ * {@link org.reactivestreams.Publisher} ends with an exception, and after that
+ * exception has been resolved to GraphQL errors.
+ *
+ * <p>The underlying transport, e.g. WebSocket, can handle a
+ * {@link SubscriptionPublisherException} and send a final error message to the
+ * client with the list of GraphQL errors.
+ *
+ * @author Mykyta Ivchenko
+ * @author Rossen Stoyanchev
+ * @since 1.0.1
+ */
+@SuppressWarnings("serial")
+public final class SubscriptionPublisherException extends NestedRuntimeException {
+
+    private final List<GraphQLError> errors;
+
+
+    /**
+     * Constructor with the resolved GraphQL errors and the original exception
+     * from the GraphQL subscription {@link org.reactivestreams.Publisher}.
+     */
+    public SubscriptionPublisherException(List<GraphQLError> errors, Throwable cause) {
+        super("GraphQL subscription ended with error(s): " + errors, cause);
+        this.errors = errors;
+    }
+
+
+    /**
+     * Return the GraphQL errors the exception was resolved to by the configured
+     * {@link SubscriptionExceptionResolver}'s. These errors can be included in
+     * an error message to be sent to the client by the underlying transport.
+     */
+    public List<GraphQLError> getErrors() {
+        return this.errors;
+    }
+
+}

spring-graphql/src/main/java/org/springframework/graphql/server/support/GraphQlWebSocketMessage.java

@@ -19,6 +19,7 @@ package org.springframework.graphql.server.support;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 import graphql.GraphQLError;
 
@@ -186,12 +187,12 @@ public class GraphQlWebSocketMessage {
 	/**
 	 * Create an {@code "error"} server message.
 	 * @param id unique request id
-	 * @param error the error to add as the message payload
+	 * @param errors the error to add as the message payload
 	 */
-	public static GraphQlWebSocketMessage error(String id, GraphQLError error) {
-		Assert.notNull(error, "GraphQlError is required");
-		List<Map<String, Object>> errors = Collections.singletonList(error.toSpecification());
-		return new GraphQlWebSocketMessage(id, GraphQlWebSocketMessageType.ERROR, errors);
+	public static GraphQlWebSocketMessage error(String id, List<GraphQLError> errors) {
+		Assert.notNull(errors, "GraphQlError's are required");
+		return new GraphQlWebSocketMessage(id, GraphQlWebSocketMessageType.ERROR,
+				errors.stream().map(GraphQLError::toSpecification).collect(Collectors.toList()));
 	}
 
 	/**

spring-graphql/src/main/java/org/springframework/graphql/server/webflux/CodecDelegate.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.graphql.server.webflux;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 
 import graphql.GraphQLError;
@@ -25,6 +27,8 @@ import org.springframework.core.codec.Decoder;
 import org.springframework.core.codec.Encoder;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.graphql.execution.ErrorType;
+import org.springframework.graphql.execution.SubscriptionPublisherException;
 import org.springframework.graphql.server.support.GraphQlWebSocketMessage;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.CodecConfigurer;
@@ -98,13 +102,17 @@ final class CodecDelegate {
 	}
 
 	public WebSocketMessage encodeError(WebSocketSession session, String id, Throwable ex) {
-		GraphQLError error = GraphqlErrorBuilder.newError().message(ex.getMessage()).build();
-		return encode(session, GraphQlWebSocketMessage.error(id, error));
+		List<GraphQLError> errors = ((ex instanceof SubscriptionPublisherException) ?
+				((SubscriptionPublisherException) ex).getErrors() :
+				Collections.singletonList(GraphqlErrorBuilder.newError()
+						.message("Subscription error")
+						.errorType(ErrorType.INTERNAL_ERROR)
+						.build()));
+		return encode(session, GraphQlWebSocketMessage.error(id, errors));
 	}
 
 	public WebSocketMessage encodeComplete(WebSocketSession session, String id) {
 		return encode(session, GraphQlWebSocketMessage.complete(id));
 	}
 
-
 }

spring-graphql/src/main/java/org/springframework/graphql/server/webflux/GraphQlWebSocketHandler.java

@@ -212,11 +212,11 @@ public class GraphQlWebSocketHandler implements WebSocketHandler {
 				.map(responseMap -> this.codecDelegate.encodeNext(session, id, responseMap))
 				.concatWith(Mono.fromCallable(() -> this.codecDelegate.encodeComplete(session, id)))
 				.onErrorResume(ex -> {
-						if (ex instanceof SubscriptionExistsException) {
-							CloseStatus status = new CloseStatus(4409, "Subscriber for " + id + " already exists");
-							return GraphQlStatus.close(session, status);
-						}
-						return Mono.fromCallable(() -> this.codecDelegate.encodeError(session, id, ex));
+					if (ex instanceof SubscriptionExistsException) {
+						CloseStatus status = new CloseStatus(4409, "Subscriber for " + id + " already exists");
+						return GraphQlStatus.close(session, status);
+					}
+					return Mono.fromCallable(() -> this.codecDelegate.encodeError(session, id, ex));
 				});
 	}
 

spring-graphql/src/main/java/org/springframework/graphql/server/webmvc/GraphQlWebSocketHandler.java

@@ -47,6 +47,8 @@ import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 
+import org.springframework.graphql.execution.ErrorType;
+import org.springframework.graphql.execution.SubscriptionPublisherException;
 import org.springframework.graphql.execution.ThreadLocalAccessor;
 import org.springframework.graphql.server.WebGraphQlHandler;
 import org.springframework.graphql.server.WebGraphQlResponse;
@@ -281,14 +283,18 @@ public class GraphQlWebSocketHandler extends TextWebSocketHandler implements Sub
 				.map(responseMap -> encode(GraphQlWebSocketMessage.next(id, responseMap)))
 				.concatWith(Mono.fromCallable(() -> encode(GraphQlWebSocketMessage.complete(id))))
 				.onErrorResume((ex) -> {
-						if (ex instanceof SubscriptionExistsException) {
-							CloseStatus status = new CloseStatus(4409, "Subscriber for " + id + " already exists");
-							GraphQlStatus.closeSession(session, status);
-							return Flux.empty();
-						}
-						String message = ex.getMessage();
-						GraphQLError error = GraphqlErrorBuilder.newError().message(message).build();
-						return Mono.just(encode(GraphQlWebSocketMessage.error(id, error)));
+					if (ex instanceof SubscriptionExistsException) {
+						CloseStatus status = new CloseStatus(4409, "Subscriber for " + id + " already exists");
+						GraphQlStatus.closeSession(session, status);
+						return Flux.empty();
+					}
+					List<GraphQLError> errors = ((ex instanceof SubscriptionPublisherException) ?
+							((SubscriptionPublisherException) ex).getErrors() :
+							Collections.singletonList(GraphqlErrorBuilder.newError()
+									.message("Subscription error")
+									.errorType(ErrorType.INTERNAL_ERROR)
+									.build()));
+					return Mono.just(encode(GraphQlWebSocketMessage.error(id, errors)));
 				});
 	}
 

spring-graphql/src/test/java/org/springframework/graphql/ResponseHelper.java

@@ -150,6 +150,9 @@ public class ResponseHelper {
 
 	public static Flux<ResponseHelper> forSubscription(ExecutionResult result) {
 		assertThat(result.getErrors()).as("Errors present in GraphQL response").isEmpty();
+		Object data = result.getData();
+		assertThat(data).as("Expected Publisher from subscription").isNotNull();
+		assertThat(data).as("Expected Publisher from subscription").isInstanceOf(Publisher.class);
 		Publisher<ExecutionResult> publisher = result.getData();
 		return Flux.from(publisher).map(ResponseHelper::forResult);
 	}

spring-graphql/src/test/java/org/springframework/graphql/client/MockGraphQlWebSocketServer.java

@@ -16,6 +16,7 @@
 
 package org.springframework.graphql.client;
 
+import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.function.Function;
@@ -29,8 +30,8 @@ import reactor.core.publisher.Mono;
 
 import org.springframework.graphql.GraphQlRequest;
 import org.springframework.graphql.GraphQlResponse;
-import org.springframework.graphql.support.DefaultGraphQlRequest;
 import org.springframework.graphql.server.support.GraphQlWebSocketMessage;
+import org.springframework.graphql.support.DefaultGraphQlRequest;
 import org.springframework.http.codec.ClientCodecConfigurer;
 import org.springframework.lang.Nullable;
 import org.springframework.web.reactive.socket.WebSocketHandler;
@@ -105,7 +106,7 @@ public final class MockGraphQlWebSocketServer implements WebSocketHandler {
 						.map(response -> GraphQlWebSocketMessage.next(id, response.toMap()))
 						.concatWithValues(
 								request.getError() != null ?
-										GraphQlWebSocketMessage.error(id, request.getError()) :
+										GraphQlWebSocketMessage.error(id, Collections.singletonList(request.getError())) :
 										GraphQlWebSocketMessage.complete(id));
 			case COMPLETE:
 				return Flux.empty();

spring-graphql/src/test/java/org/springframework/graphql/execution/CompositeSubscriptionExceptionResolverTests.java

@@ -0,0 +1,133 @@
+/*
+ * Copyright 2002-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.graphql.execution;
+
+import java.time.Duration;
+import java.util.List;
+
+import graphql.ExecutionInput;
+import graphql.GraphQL;
+import graphql.GraphQLError;
+import graphql.GraphqlErrorBuilder;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+import reactor.util.context.Context;
+import reactor.util.context.ContextView;
+
+import org.springframework.graphql.GraphQlSetup;
+import org.springframework.graphql.ResponseHelper;
+import org.springframework.graphql.TestThreadLocalAccessor;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for resolving exceptions via {@link SubscriptionExceptionResolver}.
+ * @author Rossen Stoyanchev
+ */
+public class CompositeSubscriptionExceptionResolverTests {
+
+	private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+
+	@Test
+	void subscriptionPublisherExceptionResolved() {
+		String query = "subscription { greetings }";
+		String schema = "type Subscription { greetings: String! } type Query { greeting: String! }";
+
+		GraphQL graphQL = GraphQlSetup.schemaContent(schema)
+				.subscriptionFetcher("greetings", env ->
+						Flux.create(emitter -> {
+							emitter.next("a");
+							emitter.error(new RuntimeException("Test Exception"));
+							emitter.next("b");
+						}))
+				.subscriptionExceptionResolvers(SubscriptionExceptionResolver.forSingleError(exception ->
+						GraphqlErrorBuilder.newError()
+								.message("Error: " + exception.getMessage())
+								.errorType(ErrorType.BAD_REQUEST)
+								.build()))
+				.toGraphQl();
+
+		ExecutionInput input = ExecutionInput.newExecutionInput(query).build();
+		Flux<ResponseHelper> flux = Mono.fromFuture(graphQL.executeAsync(input))
+				.map(ResponseHelper::forSubscription)
+				.block(TIMEOUT);
+
+		StepVerifier.create(flux)
+				.consumeNextWith((helper) -> assertThat(helper.toEntity("greetings", String.class)).isEqualTo("a"))
+				.consumeErrorWith((ex) -> {
+					SubscriptionPublisherException theEx = (SubscriptionPublisherException) ex;
+					List<GraphQLError> errors = theEx.getErrors();
+					assertThat(errors).hasSize(1);
+					assertThat(errors.get(0).getMessage()).isEqualTo("Error: Test Exception");
+					assertThat(errors.get(0).getErrorType()).isEqualTo(ErrorType.BAD_REQUEST);
+				})
+				.verify(TIMEOUT);
+	}
+
+	@Test
+	void resolveExceptionWithThreadLocal() {
+		String query = "subscription { greetings }";
+		String schema = "type Subscription { greetings: String! } type Query { greeting: String! }";
+
+		ThreadLocal<String> nameThreadLocal = new ThreadLocal<>();
+		nameThreadLocal.set("007");
+		TestThreadLocalAccessor<String> accessor = new TestThreadLocalAccessor<>(nameThreadLocal);
+
+		try {
+			SubscriptionExceptionResolverAdapter resolver = SubscriptionExceptionResolver.forSingleError(exception ->
+					GraphqlErrorBuilder.newError()
+							.message("Error: " + exception.getMessage() + ", name=" + nameThreadLocal.get())
+							.errorType(ErrorType.BAD_REQUEST)
+							.build());
+			resolver.setThreadLocalContextAware(true);
+
+			GraphQL graphQL = GraphQlSetup.schemaContent(schema)
+					.subscriptionFetcher("greetings", env ->
+							Flux.create(emitter -> {
+								emitter.next("a");
+								emitter.error(new RuntimeException("Test Exception"));
+							}))
+					.subscriptionExceptionResolvers(resolver)
+					.toGraphQl();
+
+			ContextView view = ReactorContextManager.extractThreadLocalValues(accessor, Context.empty());
+			ExecutionInput input = ExecutionInput.newExecutionInput(query).build();
+			ReactorContextManager.setReactorContext(view, input.getGraphQLContext());
+
+			Flux<ResponseHelper> flux = Mono.delay(Duration.ofMillis(10))
+					.flatMap((aLong) -> Mono.fromFuture(graphQL.executeAsync(input)).map(ResponseHelper::forSubscription))
+					.block(TIMEOUT);
+
+			StepVerifier.create(flux)
+					.consumeNextWith((helper) -> assertThat(helper.toEntity("greetings", String.class)).isEqualTo("a"))
+					.consumeErrorWith((ex) -> {
+						SubscriptionPublisherException theEx = (SubscriptionPublisherException) ex;
+						List<GraphQLError> errors = theEx.getErrors();
+						assertThat(errors).hasSize(1);
+						assertThat(errors.get(0).getMessage()).isEqualTo("Error: Test Exception, name=007");
+						assertThat(errors.get(0).getErrorType()).isEqualTo(ErrorType.BAD_REQUEST);
+					})
+					.verify(TIMEOUT);
+		}
+		finally {
+			nameThreadLocal.remove();
+		}
+	}
+
+}

spring-graphql/src/test/java/org/springframework/graphql/execution/ContextDataFetcherDecoratorTests.java

@@ -17,11 +17,14 @@
 package org.springframework.graphql.execution;
 
 import java.time.Duration;
+import java.util.Collections;
 import java.util.List;
 
 import graphql.ExecutionInput;
 import graphql.ExecutionResult;
 import graphql.GraphQL;
+import graphql.GraphQLError;
+import graphql.GraphqlErrorBuilder;
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -29,8 +32,8 @@ import reactor.test.StepVerifier;
 import reactor.util.context.Context;
 import reactor.util.context.ContextView;
 
-import org.springframework.graphql.ResponseHelper;
 import org.springframework.graphql.GraphQlSetup;
+import org.springframework.graphql.ResponseHelper;
 import org.springframework.graphql.TestThreadLocalAccessor;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -41,9 +44,13 @@ import static org.assertj.core.api.Assertions.assertThat;
  */
 public class ContextDataFetcherDecoratorTests {
 
+	private static final String SCHEMA_CONTENT =
+			"type Query { greeting: String, greetings: [String] } type Subscription { greetings: String }";
+
+
 	@Test
 	void monoDataFetcher() throws Exception {
-		GraphQL graphQl = GraphQlSetup.schemaContent("type Query { greeting: String }")
+		GraphQL graphQl = GraphQlSetup.schemaContent(SCHEMA_CONTENT)
 				.queryFetcher("greeting", (env) ->
 						Mono.deferContextual((context) -> {
 							Object name = context.get("name");
@@ -62,7 +69,7 @@ public class ContextDataFetcherDecoratorTests {
 
 	@Test
 	void fluxDataFetcher() throws Exception {
-		GraphQL graphQl = GraphQlSetup.schemaContent("type Query { greetings: [String] }")
+		GraphQL graphQl = GraphQlSetup.schemaContent(SCHEMA_CONTENT)
 				.queryFetcher("greetings", (env) ->
 						Mono.delay(Duration.ofMillis(50))
 								.flatMapMany((aLong) -> Flux.deferContextual((context) -> {
@@ -82,7 +89,7 @@ public class ContextDataFetcherDecoratorTests {
 
 	@Test
 	void fluxDataFetcherSubscription() throws Exception {
-		GraphQL graphQl = GraphQlSetup.schemaContent("type Query { greeting: String } type Subscription { greetings: String }")
+		GraphQL graphQl = GraphQlSetup.schemaContent(SCHEMA_CONTENT)
 				.subscriptionFetcher("greetings", (env) ->
 						Mono.delay(Duration.ofMillis(50))
 								.flatMapMany((aLong) -> Flux.deferContextual((context) -> {
@@ -104,13 +111,54 @@ public class ContextDataFetcherDecoratorTests {
 				.verifyComplete();
 	}
 
+	@Test
+	void fluxDataFetcherSubscriptionThrowException() throws Exception {
+
+		SubscriptionExceptionResolver resolver =
+				SubscriptionExceptionResolver.forSingleError(exception ->
+						GraphqlErrorBuilder.newError()
+								.message("Error: " + exception.getMessage())
+								.errorType(ErrorType.BAD_REQUEST)
+								.extensions(Collections.singletonMap("a", "b"))
+								.build());
+
+		GraphQL graphQl = GraphQlSetup.schemaContent(SCHEMA_CONTENT)
+				.subscriptionExceptionResolvers(resolver)
+				.subscriptionFetcher("greetings",
+						(env) -> Mono.delay(Duration.ofMillis(50))
+								.handle((aLong, sink) -> {
+									sink.next("Hi!");
+									sink.error(new RuntimeException("Example Error"));
+								}))
+				.toGraphQl();
+
+		String query = "subscription { greetings }";
+		ExecutionInput input = ExecutionInput.newExecutionInput().query(query).build();
+		ExecutionResult result = graphQl.executeAsync(input).get();
+
+		Flux<String> flux = ResponseHelper.forSubscription(result)
+				.map(message -> message.toEntity("greetings", String.class));
+
+		StepVerifier.create(flux)
+				.expectNext("Hi!")
+				.expectErrorSatisfies(ex -> {
+					List<GraphQLError> errors = ((SubscriptionPublisherException) ex).getErrors();
+					assertThat(errors).hasSize(1);
+					assertThat(errors.get(0).getMessage()).isEqualTo("Error: Example Error");
+					assertThat(errors.get(0).getErrorType()).isEqualTo(ErrorType.BAD_REQUEST);
+					assertThat(errors.get(0).getExtensions()).isEqualTo(Collections.singletonMap("a", "b"));
+
+				})
+				.verify();
+	}
+
 	@Test
 	void dataFetcherWithThreadLocalContext() {
 		ThreadLocal<String> nameThreadLocal = new ThreadLocal<>();
 		nameThreadLocal.set("007");
 		TestThreadLocalAccessor<String> accessor = new TestThreadLocalAccessor<>(nameThreadLocal);
 		try {
-			GraphQL graphQl = GraphQlSetup.schemaContent("type Query { greeting: String }")
+			GraphQL graphQl = GraphQlSetup.schemaContent(SCHEMA_CONTENT)
 					.queryFetcher("greeting", (env) -> "Hello " + nameThreadLocal.get())
 					.toGraphQl();
 

spring-graphql/src/test/java/org/springframework/graphql/execution/ExceptionResolversExceptionHandlerTests.java

@@ -51,7 +51,7 @@ public class ExceptionResolversExceptionHandlerTests {
 	@Test
 	void resolveException() throws Exception {
 		DataFetcherExceptionResolver resolver =
-				DataFetcherExceptionResolverAdapter.from((ex, env) ->
+				DataFetcherExceptionResolver.forSingleError((ex, env) ->
 						GraphqlErrorBuilder.newError(env)
 								.message("Resolved error: " + ex.getMessage())
 								.errorType(ErrorType.BAD_REQUEST).build());
@@ -93,7 +93,7 @@ public class ExceptionResolversExceptionHandlerTests {
 		TestThreadLocalAccessor<String> accessor = new TestThreadLocalAccessor<>(nameThreadLocal);
 		try {
 			DataFetcherExceptionResolverAdapter resolver =
-					DataFetcherExceptionResolverAdapter.from((ex, env) ->
+					DataFetcherExceptionResolver.forSingleError((ex, env) ->
 							GraphqlErrorBuilder.newError(env)
 									.message("Resolved error: " + ex.getMessage() + ", name=" + nameThreadLocal.get())
 									.errorType(ErrorType.BAD_REQUEST)
@@ -119,7 +119,7 @@ public class ExceptionResolversExceptionHandlerTests {
 	@Test
 	void unresolvedException() throws Exception {
 		DataFetcherExceptionResolverAdapter resolver =
-				DataFetcherExceptionResolverAdapter.from((ex, env) -> null);
+				DataFetcherExceptionResolver.forSingleError((ex, env) -> null);
 
 		ExecutionResult result = this.graphQlSetup.exceptionResolver(resolver).toGraphQl()
 				.executeAsync(this.input).get();

spring-graphql/src/test/java/org/springframework/graphql/server/WebGraphQlHandlerTests.java

@@ -118,10 +118,11 @@ public class WebGraphQlHandlerTests {
 		nameThreadLocal.set("007");
 		TestThreadLocalAccessor<String> threadLocalAccessor = new TestThreadLocalAccessor<>(nameThreadLocal);
 		try {
-			DataFetcherExceptionResolverAdapter exceptionResolver = DataFetcherExceptionResolverAdapter.from((ex, env) ->
-					GraphqlErrorBuilder.newError(env)
-							.message("Resolved error: " + ex.getMessage() + ", name=" + nameThreadLocal.get())
-							.errorType(ErrorType.BAD_REQUEST).build());
+			DataFetcherExceptionResolverAdapter exceptionResolver =
+					DataFetcherExceptionResolver.forSingleError((ex, env) ->
+							GraphqlErrorBuilder.newError(env)
+									.message("Resolved error: " + ex.getMessage() + ", name=" + nameThreadLocal.get())
+									.errorType(ErrorType.BAD_REQUEST).build());
 			exceptionResolver.setThreadLocalContextAware(true);
 
 			Mono<WebGraphQlResponse> responseMono = this.graphQlSetup.queryFetcher("greeting", this.errorDataFetcher)

spring-graphql/src/test/java/org/springframework/graphql/server/webflux/GraphQlWebSocketHandlerTests.java

@@ -37,11 +37,12 @@ import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.graphql.GraphQlSetup;
+import org.springframework.graphql.execution.ErrorType;
 import org.springframework.graphql.server.ConsumeOneAndNeverCompleteInterceptor;
 import org.springframework.graphql.server.WebGraphQlHandler;
 import org.springframework.graphql.server.WebGraphQlInterceptor;
-import org.springframework.graphql.server.WebSocketHandlerTestSupport;
 import org.springframework.graphql.server.WebSocketGraphQlInterceptor;
+import org.springframework.graphql.server.WebSocketHandlerTestSupport;
 import org.springframework.graphql.server.WebSocketSessionInfo;
 import org.springframework.graphql.server.support.GraphQlWebSocketMessage;
 import org.springframework.graphql.server.support.GraphQlWebSocketMessageType;
@@ -309,32 +310,25 @@ public class GraphQlWebSocketHandlerTests extends WebSocketHandlerTestSupport {
 	}
 
 	@Test
-	void errorMessagePayloadIsArray() {
-		final String GREETING_QUERY = "{" +
+	void subscriptionErrorPayloadIsArray() {
+		String query = "{" +
 				"\"id\":\"" + SUBSCRIPTION_ID + "\"," +
 				"\"type\":\"subscribe\"," +
-				"\"payload\":{\"query\": \"" +
-				"  subscription TestTypenameSubscription {" +
-				"    greeting" +
-				"  }\"}" +
+				"\"payload\":{\"query\": \"subscription { greetings }\"}" +
 				"}";
 
-		String schema = "type Subscription { greeting: String! } type Query { greetingUnused: String! }";
-
-		WebGraphQlHandler initHandler = GraphQlSetup.schemaContent(schema)
-				.subscriptionFetcher("greeting", env -> Flux.just("a", null, "b"))
-				.interceptor()
-				.toWebGraphQlHandler();
-
-		GraphQlWebSocketHandler handler = new GraphQlWebSocketHandler(
-				initHandler,
-				ServerCodecConfigurer.create(),
-				Duration.ofSeconds(60));
+		String schema = "type Subscription { greetings: String! } type Query { greeting: String! }";
 
 		TestWebSocketSession session = new TestWebSocketSession(Flux.just(
 				toWebSocketMessage("{\"type\":\"connection_init\"}"),
-				toWebSocketMessage(GREETING_QUERY)));
-		handler.handle(session).block(TIMEOUT);
+				toWebSocketMessage(query)));
+
+		WebGraphQlHandler webHandler = GraphQlSetup.schemaContent(schema)
+				.subscriptionFetcher("greetings", env -> Flux.just("a", null, "b"))
+				.toWebGraphQlHandler();
+
+		new GraphQlWebSocketHandler(webHandler, ServerCodecConfigurer.create(), TIMEOUT)
+				.handle(session).block(TIMEOUT);
 
 		StepVerifier.create(session.getOutput())
 				.consumeNextWith((message) -> assertMessageType(message, GraphQlWebSocketMessageType.CONNECTION_ACK))
@@ -343,22 +337,17 @@ public class GraphQlWebSocketHandlerTests extends WebSocketHandlerTestSupport {
 					assertThat(actual.getId()).isEqualTo(SUBSCRIPTION_ID);
 					assertThat(actual.resolvedType()).isEqualTo(GraphQlWebSocketMessageType.NEXT);
 					assertThat(actual.<Map<String, Object>>getPayload())
-							.extractingByKey("data", as(InstanceOfAssertFactories.map(String.class, Object.class)))
-							.containsEntry("greeting", "a");
+							.containsEntry("data", Collections.singletonMap("greetings", "a"));
 				})
 				.consumeNextWith((message) -> {
 					GraphQlWebSocketMessage actual = decode(message);
 					assertThat(actual.getId()).isEqualTo(SUBSCRIPTION_ID);
 					assertThat(actual.resolvedType()).isEqualTo(GraphQlWebSocketMessageType.ERROR);
-					assertThat(actual.<List<Map<String, Object>>>getPayload())
-							.asList().hasSize(1)
-							.allSatisfy(theError -> assertThat(theError)
-									.asInstanceOf(InstanceOfAssertFactories.map(String.class, Object.class))
-									.hasSize(3)
-									.hasEntrySatisfying("locations", loc -> assertThat(loc).asList().isEmpty())
-									.hasEntrySatisfying("message", msg -> assertThat(msg).asString().contains("null"))
-									.extractingByKey("extensions", as(InstanceOfAssertFactories.map(String.class, Object.class)))
-									.containsEntry("classification", "DataFetchingException"));
+					List<Map<String, Object>> errors = actual.getPayload();
+					assertThat(errors).hasSize(1);
+					assertThat(errors.get(0)).containsEntry("message", "Subscription error");
+					assertThat(errors.get(0)).containsEntry("extensions",
+							Collections.singletonMap("classification", ErrorType.INTERNAL_ERROR.name()));
 				})
 				.expectComplete()
 				.verify(TIMEOUT);

spring-graphql/src/test/java/org/springframework/graphql/server/webmvc/GraphQlWebSocketHandlerTests.java

@@ -36,6 +36,7 @@ import reactor.test.StepVerifier;
 
 import org.springframework.graphql.GraphQlSetup;
 import org.springframework.graphql.TestThreadLocalAccessor;
+import org.springframework.graphql.execution.ErrorType;
 import org.springframework.graphql.execution.ThreadLocalAccessor;
 import org.springframework.graphql.server.ConsumeOneAndNeverCompleteInterceptor;
 import org.springframework.graphql.server.WebGraphQlHandler;
@@ -323,26 +324,20 @@ public class GraphQlWebSocketHandlerTests extends WebSocketHandlerTestSupport {
 	}
 
 	@Test
-	void errorMessagePayloadIsCorrectArray() throws Exception {
+	void subscriptionErrorPayloadIsArray() throws Exception {
 		final String GREETING_QUERY = "{" +
 				"\"id\":\"" + SUBSCRIPTION_ID + "\"," +
 				"\"type\":\"subscribe\"," +
-				"\"payload\":{\"query\": \"" +
-				"  subscription TestTypenameSubscription {" +
-				"    greeting" +
-				"  }\"}" +
+				"\"payload\":{\"query\": \"subscription { greetings }\"}" +
 				"}";
 
-		String schema = "type Subscription { greeting: String! }type Query { greetingUnused: String! }";
+		String schema = "type Subscription { greetings: String! }type Query { greeting: String! }";
 
-		WebGraphQlHandler initHandler = GraphQlSetup.schemaContent(schema)
-				.subscriptionFetcher("greeting", env -> Flux.just("a", null, "b"))
-				.interceptor()
+		WebGraphQlHandler webHandler = GraphQlSetup.schemaContent(schema)
+				.subscriptionFetcher("greetings", env -> Flux.just("a", null, "b"))
 				.toWebGraphQlHandler();
 
-		GraphQlWebSocketHandler handler = new GraphQlWebSocketHandler(initHandler, converter, Duration.ofSeconds(60));
-
-		handle(handler,
+		handle(new GraphQlWebSocketHandler(webHandler, converter, TIMEOUT),
 				new TextMessage("{\"type\":\"connection_init\"}"),
 				new TextMessage(GREETING_QUERY));
 
@@ -353,22 +348,17 @@ public class GraphQlWebSocketHandlerTests extends WebSocketHandlerTestSupport {
 					assertThat(actual.getId()).isEqualTo(SUBSCRIPTION_ID);
 					assertThat(actual.resolvedType()).isEqualTo(GraphQlWebSocketMessageType.NEXT);
 					assertThat(actual.<Map<String, Object>>getPayload())
-							.extractingByKey("data", as(InstanceOfAssertFactories.map(String.class, Object.class)))
-							.containsEntry("greeting", "a");
+							.containsEntry("data", Collections.singletonMap("greetings", "a"));
 				})
 				.consumeNextWith((message) -> {
 					GraphQlWebSocketMessage actual = decode(message);
 					assertThat(actual.getId()).isEqualTo(SUBSCRIPTION_ID);
 					assertThat(actual.resolvedType()).isEqualTo(GraphQlWebSocketMessageType.ERROR);
-					assertThat(actual.<List<Map<String, Object>>>getPayload())
-							.asList().hasSize(1)
-							.allSatisfy(theError -> assertThat(theError)
-									.asInstanceOf(InstanceOfAssertFactories.map(String.class, Object.class))
-									.hasSize(3)
-									.hasEntrySatisfying("locations", loc -> assertThat(loc).asList().isEmpty())
-									.hasEntrySatisfying("message", msg -> assertThat(msg).asString().contains("null"))
-									.extractingByKey("extensions", as(InstanceOfAssertFactories.map(String.class, Object.class)))
-									.containsEntry("classification", "DataFetchingException"));
+					List<Map<String, Object>> errors = actual.getPayload();
+					assertThat(errors).hasSize(1);
+					assertThat(errors.get(0)).containsEntry("message", "Subscription error");
+					assertThat(errors.get(0)).containsEntry("extensions",
+							Collections.singletonMap("classification", ErrorType.INTERNAL_ERROR.name()));
 				})
 				.then(this.session::close)
 				.expectComplete()

spring-graphql/src/testFixtures/java/org/springframework/graphql/GraphQlSetup.java

@@ -34,6 +34,7 @@ import org.springframework.graphql.execution.DataLoaderRegistrar;
 import org.springframework.graphql.execution.DefaultExecutionGraphQlService;
 import org.springframework.graphql.execution.GraphQlSource;
 import org.springframework.graphql.execution.RuntimeWiringConfigurer;
+import org.springframework.graphql.execution.SubscriptionExceptionResolver;
 import org.springframework.graphql.execution.ThreadLocalAccessor;
 import org.springframework.graphql.server.WebGraphQlHandler;
 import org.springframework.graphql.server.WebGraphQlInterceptor;
@@ -99,6 +100,11 @@ public class GraphQlSetup implements GraphQlServiceSetup {
 		return this;
 	}
 
+	public GraphQlSetup subscriptionExceptionResolvers(SubscriptionExceptionResolver... resolvers) {
+		this.graphQlSourceBuilder.subscriptionExceptionResolvers(Arrays.asList(resolvers));
+		return this;
+	}
+
 	public GraphQlSetup typeResolver(TypeResolver typeResolver) {
 		this.graphQlSourceBuilder.defaultTypeResolver(typeResolver);
 		return this;