SAP Aggregator Pattern
The Aggregator pattern collects multiple individual messages and combines them into a single larger message before delivery. Instead of sending ten separate invoices to SAP one at a time, the Aggregator waits until all ten are available and then sends one message containing all ten. This reduces the number of transactions, improves performance, and matches the batch expectations of many SAP posting processes.
Think of the Aggregator like a bus route. A taxi takes one passenger at a time to the destination. A bus waits at the stop until it collects a full load of passengers, then makes one trip. The Aggregator is the bus — it collects messages until a trigger fires, then delivers everything together.
When to Use the Aggregator
- Multiple source events belong logically to one SAP business document (header + multiple line items arrive as separate messages)
- SAP performs better receiving one large batch than many small individual calls
- A downstream system has rate limits that make individual calls impractical
- Business rules require all related messages to arrive before processing begins
- You want to reduce IDoc or BAPI call volume during peak hours
The Aggregator Step in CPI
CPI provides a built-in Aggregator step. Drag it from the palette into your iFlow. The Aggregator holds incoming messages in a temporary store until a completion condition is met, then releases the aggregated message to the next step.
Aggregator Configuration
Three settings control when the Aggregator releases its collected messages:
Correlation Expression
Defines which messages belong together. Messages with the same correlation value get grouped into the same aggregate. Messages with different correlation values form separate aggregates running in parallel.
Example: Group by order number
Correlation Expression: ${header.OrderNumber}
Message 1: OrderNumber=PO-001, Item=Steel → Group PO-001
Message 2: OrderNumber=PO-002, Item=Bolts → Group PO-002
Message 3: OrderNumber=PO-001, Item=Nuts → Group PO-001
Message 4: OrderNumber=PO-001, Item=Washers → Group PO-001
Result: Two groups
Group PO-001: [Steel, Nuts, Washers] → released together
Group PO-002: [Bolts] → released when its condition fires
Completion Condition
Defines when the Aggregator releases the collected messages. Three options:
- CamelAggregationCompletionPredicate – A Simple expression that evaluates to true when collection is complete. Example:
${exchangeProperty.CamelAggregatedSize} == 3releases when exactly three messages are collected. - Timeout – Release after a fixed time period, regardless of how many messages arrived. Useful when you cannot predict exactly how many messages to expect.
- Size and Timeout combined – Release when N messages are collected OR after T seconds, whichever comes first. The most practical approach for production scenarios.
Aggregation Algorithm
Defines how to combine the individual messages into one aggregate:
- Combine in sequence – Appends each new message body to the growing aggregate body
- Latest message wins – The aggregate is always the most recently arrived message (useful for state updates where only the newest value matters)
- Custom (Groovy) – Write your own aggregation logic to merge messages in any way your business requires
Aggregator Example: Order Lines to IDoc
SCENARIO: A Shopify webhook fires one message per order line item.
SAP expects one IDoc with all line items in the header.
FLOW:
[Shopify: Item 1 arrives] → [Aggregator] ←┐
[Shopify: Item 2 arrives] → [Aggregator] ←┤ Wait for all items
[Shopify: Item 3 arrives] → [Aggregator] ←┘ (timeout: 30 seconds)
↓
[3 items combined into one XML]
<Order>
<Item>...Item 1...</Item>
<Item>...Item 2...</Item>
<Item>...Item 3...</Item>
</Order>
↓
[Message Mapping: to ORDERS05 IDoc]
↓
[IDoc Adapter: post to SAP]
Custom Aggregation with Groovy
The built-in combination strategies are limited. For complex merging — for example, combining a header message with multiple detail messages into a structured XML — use a Groovy-based aggregation strategy:
import com.sap.gateway.ip.core.customdev.util.Message
// This method is called for each new message that joins the aggregate
def Message aggregate(Message oldAggregate, Message newMessage) {
def newBody = newMessage.getBody(String)
if (oldAggregate == null) {
// First message: start the aggregate
newMessage.setBody("<Items>" + extractItem(newBody))
return newMessage
} else {
// Subsequent messages: append to existing aggregate
def existingBody = oldAggregate.getBody(String)
// Remove closing tag, append new item, re-add closing tag
def updatedBody = existingBody.replace("</Items>", "") +
extractItem(newBody) + "</Items>"
oldAggregate.setBody(updatedBody)
return oldAggregate
}
}
def String extractItem(String messageBody) {
// Extract the <Item> element from the message body
def start = messageBody.indexOf("<Item>")
def end = messageBody.indexOf("</Item>") + "</Item>".length()
return messageBody.substring(start, end)
}
Persistence in the Aggregator
The Aggregator stores partially collected messages in CPI's data store. This means aggregates survive iFlow restarts. If CPI restarts while three of five expected messages have arrived, the aggregate picks up where it left off when the iFlow restarts — the first three messages are not lost.
This persistence also means incomplete aggregates remain in the data store indefinitely if the remaining messages never arrive. Use the timeout completion condition to ensure aggregates eventually release even when fewer messages than expected arrive.
Monitoring Aggregator State
View the current state of all active aggregates in Monitor → Manage Stores → Data Stores. Look for entries prefixed with the Aggregator step's name. Each entry represents one aggregate currently waiting for its completion condition. If you see entries stuck for longer than expected, the remaining messages may have failed at the source — investigate why they never arrived.
Aggregator vs Gather Step
The Aggregator collects messages arriving at different times (asynchronous collection). The Gather step collects the results of a Multicast that runs multiple branches simultaneously (synchronous collection). Use Aggregator for messages arriving over time. Use Gather to wait for parallel branches inside one iFlow to complete before continuing.