GH-3504: Explain more an async error handling (#3563)

* GH-3504: Explain more an async error handling

Fixes https://github.com/spring-projects/spring-integration/issues/3504

The reaper for an aggregator (as we as a group timeout) are based on the
`TaskScheduler`.
If non-default `TaskScheduler` is used in the configuration, recommend to
supply it with a `MessagePublishingErrorHandler`

* Rewording changes for proper language

Co-authored-by: Gary Russell <grussell@vmware.com>

Co-authored-by: Gary Russell <grussell@vmware.com>

src/reference/asciidoc/aggregator.adoc

@@ -132,7 +132,6 @@ Consequently, the `Collection<Message>` variable in the POJO is cleared too, if
 If you wish to simply release that collection as-is for further processing, you must build a new `Collection` (for example, `new ArrayList<Message>(messages)`).
 Starting with version 4.3, the framework no longer copies the messages to a new collection, to avoid undesired extra object creation.
 
-
 If the `processMessageGroup` method of the `MessageGroupProcessor` returns a collection, it must be a collection of `Message<?>` objects.
 In this case, the messages are individually released.
 Prior to version 4.2, it was not possible to provide a `MessageGroupProcessor` by using XML configuration.
@@ -853,7 +852,6 @@ public interface MessageGroupCallback {
 The callback has direct access to the store and the message group so that it can manage the persistent state (for example, by entirely removing the group from the store).
 
 The `MessageGroupStore` maintains a list of these callbacks, which it applies, on demand, to all messages whose timestamps are earlier than a time supplied as a parameter (see the `registerMessageGroupExpiryCallback(..)` and `expireMessageGroups(..)` methods, described earlier).
-For more detail, see <<reaper>>.
 
 IMPORTANT: It is important not to use the same `MessageGroupStore` instance in different aggregator components, when you intend to rely on the `expireMessageGroups` functionality.
 Every `AbstractCorrelatingMessageHandler` registers its own `MessageGroupCallback` based on the `forceComplete()` callback.
@@ -895,6 +893,10 @@ In addition to the reaper, the expiry callbacks are invoked when the application
 The `AbstractCorrelatingMessageHandler` registers its own expiry callback, and this is the link with the boolean flag `send-partial-result-on-expiry` in the XML configuration of the aggregator.
 If the flag is set to `true`, then, when the expiry callback is invoked, any unmarked messages in groups that are not yet released can be sent on to the output channel.
 
+IMPORTANT: Since the `MessageGroupStoreReaper` is called from a scheduled task, and may result in the production of a message (depending on the `sendPartialResultOnExpiry` option) to a downstream integration flow, it is recommended to supply a custom `TaskScheduler` with a `MessagePublishingErrorHandler` to handler exceptions via an `errorChannel`, as it might be expected by the regular aggregator release functionality.
+The same logic applies for group timeout functionality which also relies on a `TaskScheduler`.
+See <<./error-handling.adoc#error-handling,Error Handling>> for more information.
+
 [IMPORTANT]
 =====
 When a shared `MessageStore` is used for different correlation endpoints, you must configure a proper `CorrelationStrategy` to ensure uniqueness for group IDs.
@@ -978,4 +980,4 @@ This function is used on each message added to the group and a result condition
 The `ReleaseStrategy` may consult this condition instead of iterating over all the messages in the group.
 See `GroupConditionProvider` JavaDocs and <<./message-store.adoc#message-group-condition, Message Group Condition>> for more information.
 
-See also <<./file.adoc#file-aggregator, File Aggregator>>.
+See also <<./file.adoc#file-aggregator, File Aggregator>>.

src/reference/asciidoc/configuration.adoc

@@ -139,15 +139,17 @@ To do so, define a bean with the appropriate JNDI name for your environment, as
 ====
 [source,xml]
 ----
-<bean id="taskScheduler" class="o.s.scheduling.commonj.TimerManagerTaskScheduler">
-    <property name="timerManagerName" value="tm/MyTimerManager" />
+<bean id="taskScheduler" class="org.springframework.scheduling.concurrent.DefaultManagedTaskScheduler">
+    <property name="jndiName" value="tm/MyTimerManager" />
     <property name="resourceRef" value="true" />
 </bean>
 ----
 ====
 =====
 
-The next section describes what happens if exceptions occur within the asynchronous invocations.
+IMPORTANT: When a custom `TaskScheduler` is configured in the application context (like the above mentioned `DefaultManagedTaskScheduler`), it is recommended to supply it with a `MessagePublishingErrorHandler` (`integrationMessagePublishingErrorHandler` bean) to be able to handle exceptions as `ErrorMessage`s sent to the error channel, as is done with the default `TaskScheduler` bean provided by the framework.
+
+See also <<./error-handling.adoc#error-handling,Error Handling>> for more information.
 
 [[global-properties]]
 === Global Properties
@@ -631,7 +633,7 @@ public class MyFlowConfiguration {
 ====
 
 Version 5.0 introduced support for a `@Bean` annotated with `@InboundChannelAdapter` that returns `java.util.function.Supplier`, which can produce either a POJO or a `Message`.
-The followig example shows how to use that combination:
+The following example shows how to use that combination:
 
 ====
 [source,java]

src/reference/asciidoc/error-handling.adoc

@@ -34,7 +34,16 @@ However, you can define your own if you want to control the settings.
 The following example shows how to define an error channel in XML configuration backed by a queue with a capacity of `500`:
 
 ====
-[source,xml]
+[source, java, role="primary"]
+.Java
+----
+@Bean
+QueueChannel errorChannel() {
+    return new QueueChannel(500);
+}
+----
+[source, xml, role="secondary"]
+.XML
 ----
 <int:channel id="errorChannel">
     <int:queue capacity="500"/>
@@ -48,6 +57,9 @@ The most important thing to understand here is that the messaging-based error ha
 This does not apply to exceptions thrown by a handler that operates within the same thread as the sender (for example, through a `DirectChannel` as described earlier in this section).
 
 NOTE: When exceptions occur in a scheduled poller task's execution, those exceptions are wrapped in `ErrorMessage` instances and sent to the 'errorChannel' as well.
+This is done via a `MessagePublishingErrorHandler` injected into the global `taskScheduler` bean.
+It is recommended to use that `MessagePublishingErrorHandler` for any custom `taskScheduler` if the error handling still has to be done using standard 'errorChannel' integration flow logic.
+A registered `integrationMessagePublishingErrorHandler` bean can be used in this case.
 
 To enable global error handling, register a handler on that channel.
 For example, you can configure Spring Integration's `ErrorMessageExceptionTypeRouter` as the handler of an endpoint that is subscribed to the 'errorChannel'.