Design A Flow

A flow is a message processing pipeline that acts as the integration layer between systems. Refer to the Core Concepts for a conceptual overview on flows and their configuration options.

Flow configurations are created with the flowConfig builder in the following manner:

flowConfig {
    id = "my-flow"
    description = "My First Flow"
    ownerId = "myCompany"
    exchangePattern = RequestResponse

    restApi {
        id = "echo-api"
        apiSpecId = "OpenAPIv3:echo-api:latest"
    }

    ...
}

The first processor added to a flowConfig builder is the "source," or entry point, for that flow. The source processor will acquire or receive messages. For example, an initial restApi processor signifies that the flow will receive messages on a REST endpoint. Additional processors added to the flowConfig make up the flow pipeline, where the message processing occurs. The pipeline implements processors in the order that they are added to the flowConfig.

See Get started for tutorials on writing (and testing) example flow configurations.

Properties

The following properties can be applied to a flow configuration:

Property Description

id

The unique ID of the flow. Required.

description

Description of the flow. Optional.

ownerId

ID of the customer organization that owns this flow. ID is provided by Greenbird. Required.

exchangePattern

The processing behavior of the flow. Can be either RequestResponse, OneWay, or Headless. Required.

deliveryGuarantee

Specifies which system owns the delivery guarantee for a flow. Can be ByFlowServer or BySourceSystem.

maxNumberOfInFlightMessages

The maximum number of messages that can be processed concurrently by the pipeline. Once this limit is reached, the flow source stops sending messages to the pipeline but continues accepting and persisting them.

maxBufferSize

The maximum number of (persisted) messages held in memory at once, waiting for in-flight slots to open. Also used as the batch size when reading messages from persistence to memory.

maxStashSize

The maximum number of (persisted) messages allowed to exist on disk awaiting buffer space. Once this limit is exceeded, the source rejects new messages with an error. Note that the message remains persisted even as it moves from stash to buffer to in-flight.

correlationIdExpr

A SpEL expression to extract a correlation ID from the incoming requests in the flow specification. The correlation ID will be stored in the messageCorrelationId exchange property but won’t overwrite an existing ID/exchange property.

Exchange Patterns

The exchangePattern property determines the overall behavior of the flow and its relationship to other flows or clients. This property is set by using the FlowExchangePattern enum (e.g., FlowExchangePattern.RequestResponse). The following exchange patterns are available:

Pattern Description

RequestResponse

The traditional synchronous request, where the result of the processing is returned to the client in a response at the end of the processing.

OneWay

The pattern for asynchronous processing where the client is only concerned with delivering a message and does not wish to wait for the result of the processing. The client will receive a response/receipt immediately after the message has been reliably persisted, regardless of when the actual pipeline processing takes place. Also note that OneWay messages are still subject to delivery guarantees (i.e., redelivery and dead letter strategies).

Headless

The flow behaves more like an include statement. The processor pipeline of a Headless flow is included in the upstream pipeline at the point where it is referenced. The message processing in the Headless flow pipeline becomes subject to the exchange pattern and message reliability configurations of the upstream flow.

Delivery Guarantee

The deliveryGuarantee property determines which system owns delivery guarantee for a flow.

Value Description

ByFlowServer

Default. The flow server persists each message before processing. This mode provides a server-side at-least-once delivery guarantee, including crash recovery and redelivery.

BySourceSystem

The source system, for example Kafka, owns the delivery guarantee. The flow server does not persist messages, which reduces persistence overhead for high-throughput use cases. The source system redelivers unacknowledged messages. This option is supported only for selected source processors and detailed in the source processor documentation. It requires the RequestResponse exchange pattern and does not support throttling properties such as maxBufferSize, maxStashSize, or side channeling.

Throttling

GridOS Connect persists every incoming message in reliable storage before pipeline processing starts. During delivery, a message can be in one of the following states:

  • In-flight — persisted and currently being processed (active work in the pipeline).

  • Buffered — persisted and held in memory, waiting for an in-flight slot to become available.

  • Stashed — persisted to disk but not held in memory, waiting for buffer space to be freed.

The three throttling properties control how many messages occupy each state and when backpressure is applied: maxNumberOfInFlightMessages, maxBufferSize, and maxStashSize. The number of in-memory messages at any time is bounded by maxBufferSize + maxNumberOfInFlightMessages.

General Configuration Guidelines

The default goal for most integrations is safe, steady processing, not maximum throughput. In practice, conservative throttling settings are the best fit for most flows. Only optimize for high concurrency or fire-hose style throughput when you have explicit requirements and have validated the impact on downstream systems, memory, and persistence capacity.

Start with reasonable defaults:

  • maxNumberOfInFlightMessages: 1-10 — limits concurrent work to avoid overwhelming downstream services.

  • maxBufferSize: 10–20 — keeps a small window of messages ready to process without excessive memory.

  • maxStashSize — set based on exchange pattern and outage budget (not a fixed default range).

maxStashSize has different meaning depending on exchangePattern.

  • RequestResponse flows: Long stashing is usually not useful because clients are waiting for a response. For these flows, keep maxStashSize modest and focus on realistic response-time expectations.

  • OneWay flows: Use maxStashSize as outage protection. If the target system is down, Connect should continue accepting and persisting messages until recovery. Therefore, a large maxStashSize is often needed, sized according to expected traffic and outage tolerance.

Large stash values are generally safe from a memory perspective, but they still consume persistence storage. Validate disk capacity and operational retention expectations.

Key principles:

  • A small maxNumberOfInFlightMessages is usually sufficient to achieve high throughput, plus under clustered deployment, multiple flow instances can process messages in parallel. Increase only if you know both producer and consumer can handle more concurrency and the bottleneck is not downstream.

  • Keep maxBufferSize small to prevent one flow from starving others of memory, especially for flows that handle large messages.

  • maxStashSize is disk-backed, not memory-backed. Do not cap it aggressively by default. Set it according to how long your downstream can be unavailable and how much incoming traffic you need to absorb.

  • Monitor the Connect Dashboard in-flight, buffered, and stashed message panels to observe actual usage patterns. Adjust settings based on what you observe, not just what you predict.

Common Scenarios

Scenario 1: Fast producer, slow consumer (back-pressure case)

Source system sends messages rapidly to a flow that communicates with a slow destination API (e.g., database writes). To avoid overwhelming the destination while still accepting producer traffic, configure the flow with the following properties:

exchangePattern = OneWay              // Producer doesn't wait for response, so stashing is useful for outage protection
maxNumberOfInFlightMessages = 1       // Limit concurrency to avoid overwhelming the database
maxBufferSize = 10                    // Keep a small buffer to limit memory pressure while reducing disk-read frequency.
maxStashSize = 3000000                // Large stash > peak incoming message rate (120 msg/s) × outage window (6 hours)

Scenario 2: Large messages (memory-constrained)

A flow processes large payloads (e.g., multi-megabyte files) and must minimize memory footprint.

maxNumberOfInFlightMessages = 1        // Process one message at a time
maxBufferSize = 0                     // Do not hold messages in memory; fetch one-by-one from disk

With maxBufferSize = 0, the stash effectively becomes the queue, and messages are read from disk one at a time. This minimizes memory use at the cost of more frequent disk I/O.

Scenario 3: Small, rapid messages

A flow is processing large volumes of small messages (e.g., IoT sensor data) where the consumer can keep up with the fast producer.

maxNumberOfInFlightMessages = 10     // Allow many parallel operations
maxBufferSize = 100                  // Larger buffer for small messages (memory cost is low)

Since messages are small, higher buffering is affordable and reduces stash churn.

Special Cases

Troubleshooting High Stash Usage

If you observe the stash filling up during normal operation, use this diagnostic sequence:

  1. Verify the actual bottleneck. Is the downstream service truly slow, or is it a configuration issue (e.g., too-long redeliveryMillis, incorrect timeouts)?

  2. Adjust throttling conservatively. Increase maxNumberOfInFlightMessages or maxBufferSize only if you’ve confirmed the bottleneck is not the downstream sink. These changes increase memory use.

  3. Scale the sink if needed. If the downstream service is the confirmed bottleneck, increase capacity there instead of expanding flow buffers. This is more efficient and safer than unbounded tuning.

Splitting Large Data For Concurrent Processing

Large payloads (files, documents, or arrays) consume memory and reduce retry efficiency. Split them early and process fragments concurrently:

  1. Large file ingestion:

    • Use readFiles with small maxNumberOfInFlightMessages (1–2) and maxBufferSize = 0 to avoid holding entire files in memory.

    • Configure automatic splitting: Connect supports splitting comma-separated and line-separated files via readFiles split options (see lineSplitMaxBytes and related configuration).

    • Alternatively, split files manually before ingestion.

  2. Large payloads and arrays:

    • For delimited text and CSV data, use readFiles or parsing processors such as parseCsv and parseHierarchicalCsv to extract individual records early.

    • Use split to break large structures (arrays, documents) into smaller fragments.

    • Then, in the downstream flow, increase concurrencyMaxNo for relevant processors to process fragments in parallel.

This pattern improves memory efficiency, reduces message retry scope, and usually yields better overall throughput and stability than processing very large payloads intact.

Achieving Parallelism with Singleton Sources

Scaling flow server replicas does not automatically increase source parallelism for singleton sources.

For sources that are singleton or run in singleton mode, use manual sharding: deploy multiple flows, each handling a subset of input. For example:

  • Kafka/Event Hubs: Assign each flow instance to consume from a different partition or topic/group combination.

  • File ingestion: Deploy multiple flows where each readFiles instance reads from a different directory or file-name pattern.

Keep flow pipelines clustered unless you explicitly need singleton pipeline behavior, see Flow Pipeline Deployment.

Groups

Flows can optionally be assigned one or more group identifies. This is useful for grouping flows in Utilihive Heartbeat. Flows are assigned groups via the group() function. For example:

flowConfig {
    id = "my-flow"
    description = "My First Flow"
    ownerId = "myCompany"
    exchangePattern = RequestResponse

    group("Test Group 1")
    group("Test Group 2")

    ...
}

Strategies

Strategy builders can be added before or after the list of processors in the flowConfig to configure specifics on how a flow is deployed, archives messages, or handles failures.

GridOS Connect provides at-least-once delivery guarantee for flow sources through two complementary mechanisms:

  • redeliveryStrategy for retry behavior.

  • deadLetterStrategy for handling messages that still fail after retries.

At-least-once delivery means duplicates are possible. Downstream processing should therefore be idempotent, and strict message ordering is not guaranteed.

Failures in Connect are categorized as either transient or non-transient.

  • A transient failure may resolve on retry (for example, timeout or temporary overload).

  • A non-transient failure is not expected to resolve with retry (for example, invalid payload or incorrect credentials).

For HTTP request processors such as restRequest and soapRequest, transience is determined by HTTP status codes.

Redelivery

The redeliveryStrategy configures the at-least-once delivery contract for a flow. In other words, you can specify how often the flow source tries to redeliver a failed message before giving up and delegating the message to error handling. The strategy can be configured with the following optional properties:

Property Description

redeliveryMillis

The approximate time budget (in milliseconds) the source uses for each delivery attempt before deciding that attempt has not completed and scheduling redelivery or giving up. Set this to at least the sum of relevant downstream timeout settings plus an additional safety buffer. When redeliveryMaxNo is 0, this value still applies to the initial delivery attempt.

redeliveryMaxNo

The number of times the message will be redelivered before the source gives up and delegates the message to error handling. Setting this property to 0 will turn off redelivery attempts, but the original message is still sent for the first time.
Configuring redeliveries is strongly encouraged. Redeliveries are an integral part of ensuring reliable message processing for flows.

How To Calculate redeliveryMillis

Use this baseline formula:

redeliveryMillis >= sum(all relevant downstream timeouts) + safety buffer

Include timeout values for processors that can block completion of the delivery attempt (for example restRequest, soapRequest, database calls, external queue operations, and any custom processor timeouts).

Worked example:

  • restRequest.connectionTimeoutMillis = 5000

  • restRequest.receiveTimeoutMillis = 20000

  • Additional downstream timeout (for example another blocking call) = 3000

  • Safety buffer = 2000

Calculation:

redeliveryMillis >= 5000 + 20000 + 3000 + 2000 = 30000

In this case, start with redeliveryMillis = 30000 (30 seconds), then tune with production observations.

Defaults And When To Override

Default redelivery values are:

  • redeliveryMaxNo = 5

  • redeliveryMillis = 15000

These defaults are often too small for flows with slow downstream dependencies. If downstream timeouts already exceed 15 seconds, increase redeliveryMillis immediately.

Redelivery And In-Flight Capacity

Messages that are being redelivered continue to occupy maxNumberOfInFlightMessages slots until they either succeed or are delegated to dead letter handling.

Example:

  • maxNumberOfInFlightMessages = 10

  • 5 messages are currently in redelivery attempts

Only 5 in-flight slots remain for new messages. Effective throughput drops during failure periods, and buffered/stashed queues may grow.

This is why redelivery settings and throttling settings must be tuned together (redeliveryMaxNo, redeliveryMillis, maxNumberOfInFlightMessages, maxBufferSize, and maxStashSize).

Troubleshooting "Redelivery Attempts Exhausted"

The "redelivery attempts exhausted" error can represent two different situations:

  • Real downstream failure: every attempt failed with an actual processor error.

  • Insufficient redeliveryMillis: the source gave up waiting before processing could complete.

To differentiate them, inspect dead letter payload fields (especially failureDetails) and logs/traces:

  • If you see concrete downstream error details (for example HTTP status codes, connection errors, authentication failures), treat it as a real downstream failure.

  • If failure context repeatedly lacks specific downstream error details, suspect that redeliveryMillis is too short and increase it based on downstream timeout totals plus buffer.

Dead Letter

The deadLetterStrategy defines what happens when a flow’s configured redeliveryStrategy is exhausted, and further delivery attempts are cancelled. The strategy can be configured with the following optional properties:

Property Description

deadLetterFlowId

Flow to which the dead letter payload is forwarded and where the errors can be handled.

forwardNonMessageRelatedErrors

Whether errors that are not directly related to messages are sent to the deadLetterFlowId. An example error would be a poller issue like losing connection to the external queue.

forwardMessagePreprocessingErrors

Whether errors encountered by the source endpoint before the message is processed by the flow are sent to the deadLetterFlowId. An example error would be malformed or invalid JSON in a REST source.

Dead letter payloads include the following fields:

Field Description

originalMessagePayload

Payload of the original message.

originalMessageId

ID of the message forwarded to the dead letter flow.

originalDeliveryId

Delivery ID of the message forwarded to the dead letter flow.

failureTransience

Information on whether the error was caused by a transient or non-transient failure.

messageDeliveryAttempts

Number of attempts to deliver the message.

payloadOnFailure

Payload that was being processed on failure.

failureDetails

Object containing info about the failure (e.g., a restRequest failure would include the URL, status code, error message, etc.).

Catch and Return

The catchAndReturnStrategy is used to catch failures and produce an alternative result. This is only relevant for RequestResponse flows and can still be combined with a deadLetterStrategy. The strategy is configured by providing a processor that will receive the failure message (in the same format as a dead letter payload). The results of the processor become the response for the flow. For example:

catchAndReturnStrategy {
    processor<MapConfig.Builder> {
        id = "return-error"
        mapSpec = """
            {
                "message" : #input.payload.failureDetails.errorDetails.errorMessage
            }
        """.trimIndent()
    }
}

Note that the processor is implemented by referencing its underlying config class (e.g., MapConfig.Builder) and not the builder shorthand (e.g., map). Most processors are supported except for those that are dependent on some form of persistence (e.g., the suspendMessage processor) or multi-flow handling (e.g., distribute, split).

When executing a processor with the catchAndReturnStrategy applied, standard logs and tracing functionalities are currently unavailable for functional tests.

Flow Pipeline Deployment

The flowPipelineDeploymentStrategy determines how a flow pipeline is deployed in the flow server cluster. The flow pipeline is a separate deployment from the flow source and is either deployed with one instance on each cluster member or as a cluster singleton. The following table highlights the differences:

Strategy Description

CLUSTERED

There is a source and a pipeline on each cluster member. The message will not be remoted to another node in the cluster for pipeline processing but processed locally on the node where the source resides. This is the default strategy if the pipeline does not contain a persistent processor.

SINGLETON

Only one copy of the pipeline is deployed, and all clustered sources will send messages to the same, single pipeline. This will involve remoting for the sources that are on different nodes than the pipeline. This is the default strategy if the pipeline contains one or more persistent processors.

The default strategy cannot be overridden from SINGLETON to CLUSTERED, only CLUSTERED to SINGLETON. Thus, you would only need to implement a flowPipelineDeploymentStrategy in the case where you want a normally clustered flow to be singleton instead (like wanting messages from clustered sources to be processed in a single pipeline instance). To do so, use the scope property in the following manner:

flowPipelineDeploymentStrategy {
    scope = FlowPipelineDeploymentScope.SINGLETON
}
A singleton flow source doesn’t necessarily mean a singleton deployment. By default, the pipeline of a singleton source (such as a file-poller) will be deployed alongside the source on the same node, meaning it works similar to the clustered variant but with only one local deployment.

Response Payload Archiving

The responsePayloadArchivingStrategy is used to archive a flow’s response by sending the final payload to another flow. The receiving flow must follow handoff conventions. Each archived payload must have a unique path that can later be used to retrieve it from the archive. The path is generated using the flow and message IDs unless otherwise specified.

The strategy is configured with the following properties:

Property Description

flowId

Flow to which the payload will be forwarded. Required.

objectPathExpr

Optional expression that determines the unique path of the payload in the archive.
Individual processors, including the source processor, can be configured to archive their incoming payload with a payloadArchivingStrategy.