diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index 0b94b694eb..e9527d5e8a 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -132,7 +132,6 @@ Consequently, the `Collection` 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(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 <>. 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>>. \ No newline at end of file +See also <<./file.adoc#file-aggregator, File Aggregator>>. diff --git a/src/reference/asciidoc/configuration.adoc b/src/reference/asciidoc/configuration.adoc index 99620000a0..42116de520 100644 --- a/src/reference/asciidoc/configuration.adoc +++ b/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] ---- - - + + ---- ==== ===== -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] diff --git a/src/reference/asciidoc/error-handling.adoc b/src/reference/asciidoc/error-handling.adoc index 3efdad71cf..b29fe490f2 100644 --- a/src/reference/asciidoc/error-handling.adoc +++ b/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 ---- @@ -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'.