Archive and Resubmission

Use the archive and resubmission pattern to archive payloads at defined points in a flow for later retrieval. Operators can retrieve archived payloads and trigger resubmission from Flow Traces in the Connect Console. Use this pattern to support durable payload storage and recover messages when a backend failure lasts longer than the flow source’s redelivery cycle.
Use a payloadArchivingStrategy to archive a processor’s input, including a source processor’s input, or use a flow-level responsePayloadArchivingStrategy to archive the final response payload in a RequestResponse flow.

The pattern uses three flows with distinct roles:

Flow Role Flow ID

Archiver flow

Uses a handoff source in a OneWay flow to receive a payload copy and write the incoming payload to object storage.

Set by the flowId property in the archiving strategy; the ID is not fixed.

Archive flow

Reads an archived payload from storage using archiveLookup.

Must be exactly archive.

Resubmits flow

Receives resubmission requests and forwards eligible payloads to destination flows.

Must be exactly resubmits.
Archiving sends a copy of the payload to a dedicated handoff flow, which adds message persistence and writes the payload to object storage. This is a design trade-off: proper archiving enables reliable payload storage while supporting later retrieval and resubmission.

Archive Payloads

Archiving sends a copy of the payload to a dedicated handoff flow. The archiver flow then stores the payload in a storage service, such as a Google Cloud Storage bucket, by using the writeToGoogleStorage processor. When the archiver flow receives the handoff payload, it contains two properties: archivePath, the generated or custom storage path, and originalMessagePayload, the original payload to store. Thus, an archiver flow might look like the following example:

flowConfig {
    id = "my-archiver-handoff-flow"
    ownerId = OWNER_ID
    exchangePattern = OneWay

    handoff { id = "archiver-handoff" }

    setExchangeProperty {
        id = "set-path-as-filename"
        propertyName = "fileName"
        source = "envelope.payload.archivePath"
    }

    setPayload {
        id = "set-original-as-payload"
        source = "envelope.payload.originalMessagePayload"
    }

    writeToGoogleStorage {
        ...
    }
}

Each archived payload needs a unique path so the archive flow can retrieve it from the storage solution later. By default, the path is generated using the flow and message IDs and available in the archiver flow as the payload property archivePath. If a custom path is needed, the standard path can be overridden by setting the objectPathExpr property on the archiving strategy. For example:

payloadArchivingStrategy {
    flowId = "my-archiver-handoff-flow"
    objectPathExpr = "'my-own-id/' + envelope.originalMessageId"
}

Payload Retrieval and Resubmission

Dedicated Service Account

Create a service account named insights-archive-user of type API key.

Save the generated API key secret for future use.

Retrieve Payloads

The retrieval flow must have the exact ID archive so Connect can locate it automatically.

To retrieve payloads from the archive, create a separate flow with the ID archive and a source processor of archiveLookup. The archiveLookup processor creates a REST endpoint that receives the archive path on the property payloadArchivePath, which is then used to look up the archived payload from the given storage solution. For example:

flowConfig {
    id = "archive"
    ownerId = OWNER_ID
    exchangePattern = RequestResponse

    archiveLookup { id = "archive-retriever-api" }

    setPayload {
        id = "set-path-as-payload"
        source = "envelope.payload.payloadArchivePath"
    }

    lookupFromGoogleStorage {
        ...
    }
}

Trigger the archive flow with a GET request to /{ownerId}/rs/archive/payloads/<archive path>. When you view payloads on the Flow Traces page, Utilihive retrieves them automatically if payload retrieval is enabled for your organization and the dedicated insights-archive-user service account has access to the archive flow and all downstream flows.

Resubmit Payloads

You can resend a message after a flow processes it. For example, a backend might reject messages for longer than the flow’s redelivery cycle, causing the flow source to discard them. When the backend has been fixed, the messages can be retrieved from the archive flow and sent again. This process is called resubmission.

You can resubmit some types of archived messages from the Flow Traces page in Utilihive after you complete the following setup:

The resubmission flow must have the exact ID resubmits so Connect Console can locate it and post resubmission requests to it.

The resubmits flow must use a resubmit source. See Set Up Flows for implementation details.

The flow that receives the resubmitted message is the destination flow. A message is eligible for resubmission only when both of these conditions are met:

  • The destination flow uses a handoff source. Resubmission requires an independent inbound endpoint with its own message persistence and delivery guarantee; only a handoff source provides these. Headless flows have no independent source and cannot receive messages directly. Other source types, such as restApi and readFiles, are not supported.

  • The message was archived from the destination flow’s source. It must be identical to the message originally received by the flow, not a version archived from a later processor in the flow.

Set Up Flows

The resubmits flow is the endpoint where Utilihive Console will post resubmission requests. You can also post requests there manually if you authenticate the request with the same insights-archive-user credentials.

The flow exposes a POST endpoint that receives a list of archived paths to be resubmitted to a destination flow. For example:

{
  "payloadArchivePaths": [
    "/path/to/payload1",
    "/path/to/payload2"
  ],
  "destinationFlowId": "destination-flow-a-handoff"
}

The flow must fetch and forward messages to each destination flow that your organization supports for resubmission. The following example shows a typical resubmits flow:

flowConfig {
    id = "resubmits"
    ownerId = OWNER_ID
    exchangePattern = OneWay

    resubmit { id = "resubmit-api" }

    (1)
    map {
        id = "build-list"
        mapSpec = """
            let #archivePaths := for #path in #input.payload.payloadArchivePaths[]
            return {
                "payloadArchivePath" : #path,
                "destinationFlowId" : #input.payload.destinationFlowId
            }

            return #archivePaths
        """.trimIndent()
    }

    (2)
    split {
        id = "split-list"
        flowId = "retrieve-and-resubmit-flow"
    }
}
1 Convert the resubmit object into a list of path/destination objects using map.
2 split the list into individual messages and forward them to a handoff flow so that they can be processed individually.

The following example shows the handoff flow:

flowConfig {
    id = "retrieve-and-resubmit-flow"
    ownerId = OWNER_ID
    exchangePattern = OneWay

    handoff { id = "retrieval-handoff" }

    (1)
    saveToStash {
        id = "stash-destination"
        key = "destinationFlowId"
        source = "envelope.payload.destinationFlowId"
    }

    (2)
    lookupFromGoogleStorage {
        id = "fetch-payload"
        ...
    }

    (3)
    distribute {
        id = "distribute-for-resubmission"
        flowId("destination-flow-a-handoff")
        flowId("destination-flow-b-handoff")
        flowId("destination-flow-c-handoff")

        targetFilterExpression("targetFlowId == envelope.stash.destinationFlowId")
    }
}
1 In the handoff flow, store the destination ID for use in the distribution filter.
2 Fetch the payload to resubmit.
3 Use targetFilterExpression to route the message to the destination flow that matches destinationFlowId.