A flow subscribes to a topic by starting with a topic listener, and then runs whenever messages are published to that topic. You choose the topic, decide where a brand-new subscription starts reading, and tune how messages are grouped into pages. This article covers creating a subscription, the Start from setting, batching, how a subscription recovers its position after downtime, and how to replay retained messages.
Several flows can subscribe to the same topic, and each receives every message independently.
Prerequisites:
- The Topics license must be enabled for your account.
- You need access to at least one topic. See Create and manage topics.
Subscribe a flow to a topic
- Open the flow in the flow builder.
- Add a source step.
- Select Celigo Topics.
- Select Listen for messages from a topic.
- In the Subscription section, select a Topic.
- Set Start from if this is a new subscription. See Choose where the subscription starts.
- Enter a Name for the listener.
- Enable the flow.
The flow now runs whenever messages arrive.
Note: The listener is the subscription, and it has no Connection field. See Flow steps don't use a connection.
Choose a topic
Select a topic in the Topic picker, or select Create topic in the picker to create one without leaving the flow. See Automatic registration.
You can change the topic on a saved listener. When you do, the platform stops reading the old topic and begins reading the new one, and the subscription places itself according to its Start from and cursor rules.
Choose where the subscription starts
Start from decides where a brand-new subscription begins reading. It has two options:
| Option | Behavior |
|---|---|
| Only new messages (default) | The flow starts from the moment it is first enabled. Anything already in the topic is skipped. |
| All retained messages | The flow reads everything currently in retention, then continues with new messages. |
Use All retained messages when a new consumer needs to catch up on history — for example, when you add a flow to an existing topic and it needs the backlog. Use Only new messages when history is irrelevant or would cause duplicate downstream work.
Three things to know:
It resolves at first activation, not at save. A listener you save today and enable in three weeks starts from the moment you enable it, not from when you saved it.
It applies only once. Later disables and re-enables never re-apply it — the flow resumes from its cursor.
It disappears after the first run. Once the subscription has a position, the setting no longer applies, so it's removed from the form.
Warning: Before selecting All retained messages, consider how much data that is. Nothing tells you how many messages are currently in retention, and a topic with 90-day retention and high volume can deliver a very large backlog into your flow at once.
There is no option to start a new subscription from a specific time. To reprocess messages from a specific time after the flow has run, replay them.
Tune how messages are batched
Messages reach the flow in pages, not one at a time. Two settings in the Advanced section control how a page is assembled, and a page is delivered when either is reached first.
Page size — the maximum number of messages per page. Default: 20.
Max wait time — how many seconds to wait for a page to fill before delivering a partial one. Default: 30.
How they interact:
- Under sustained volume, pages fill quickly and Max wait time rarely matters.
- For sparse traffic, Max wait time is the delivery latency of a single message. With the default 30-second max wait, a lone message reaches the flow about 30 seconds later.
Lower Max wait time when messages are infrequent and you want them acted on quickly. Raise Page size when your subscribing flow calls a downstream API and you want fewer, larger calls.
These are the same two fields, in the same section, as on the MongoDB CDC listener.
Delivery behavior
Delivery is push. There is no schedule to configure and no manual run. The flow runs when messages arrive.
Every subscription gets every message. Reading a message never removes it. 10 subscribed flows each receive their own copy.
Each message enters the flow as one record — the payload. Records use the message ID as their trace key by default, so every record always has one. To trace by a business identifier instead, set Override trace key template on the listener to derive it from the payload.
Delivery is at least once. A message can be delivered more than once, so build subscribing flows to tolerate duplicates. See Delivery guarantees.
Design against real payloads
The Mock output panel lets you build the rest of the flow before enabling it. Set Populate with to:
- Live data — the topic's most recent retained messages, up to 10, shaped as the flow's input.
- Sample data — a generic stub.
Use Live data when the topic already has messages, so that your mappings are built against payloads that actually exist.
Pause and resume a subscription
To pause a subscription, disable the flow. To resume it, enable the flow. There is no separate setting for the subscription.
A disabled flow's subscription holds its position. When you re-enable it, it resumes from where it stopped and processes everything published in the meantime — as long as those messages are still within the topic's retention window.
Warning: Messages that age out while a flow is disabled are gone, so set retention longer than your worst-case downtime. See Choose a retention window.
How a subscription decides where to resume
A subscription's position in the topic (its cursor) is runtime state: you don't set or edit it. It advances as the flow processes messages, and replay never moves it.
| Situation | Where the flow resumes |
|---|---|
| The subscription has a position | From that position |
| The subscription has never run | Per its Start from setting |
| The stored position is no longer valid | From the earliest retained message |
A stored position can become invalid if it aged out of retention, if the topic was purged, or after internal maintenance. In every case the platform resumes from the earliest retained message rather than failing.
This deliberately favors redelivering messages over silently skipping them. A subscription never gets stuck on its own position, and you never need to repair one manually.
Replay messages
Replay redelivers messages that are still in the topic to this flow. Use it to recover after a downstream outage, a bug, or a mapping mistake, without cloning the flow or changing its subscription. You can replay only messages within the topic's retention period.
Warning: Replay redelivers messages even if the flow already processed them, so make sure the flow can process the same message twice with the same result. See Delivery guarantees.
To replay messages:
- Open the flow in the flow builder.
- Select Replay. If the flow also has batch exports, select Run, and then select Replay.
- If the flow has more than one topic listener, under Replay multiple sources, select All sources or Selected sources. Each source replays from its own starting point.
- Under Replay setting, select where to start:
- Earliest retained — replays every message the topic still holds.
- Specific time — enter a start date, time, and timezone. Messages published from that time forward are replayed.
- Optional: to replay only a known window, select Include end date and time (optional), then enter the end date, time, and timezone. The end must be after the start and can't be later than now.
- Select Start.
Note: If any step uses an offline connection, the replay dialog warns that those steps won't complete. The warning doesn't block the replay. To complete every step, bring the connections online first.
The replay runs as its own job (Waiting in queue, then Run in progress) and completes when every message in the range has been delivered. New messages keep arriving through the flow's normal subscription while the replay runs.
What to know about replay
- Only this flow is affected. Other flows subscribed to the topic, including flows that share this flow's listener, don't receive the replayed messages.
- The subscription's position doesn't change. After the replay completes, the flow continues from where it was.
- Starting before the retention period is fine. If the start time is older than the oldest retained message, the replay begins at the oldest retained message.
- Earliest retained doesn't show a date. To see how far back the topic goes, check its message history. See View and purge a topic's message history.
- Replay is available only when the flow is turned on and not in test mode. Anyone who can run the flow can replay it, including monitor users.
- You can replay more sources while a replay runs. Select Replay additional sources. In a flow that also has batch exports, select Run additional sources, and then select Replay. Select at least one source. When every topic source is replaying, no replay action is available.
- Replay is only in the flow builder. The Run action on the integration's Flows tab doesn't offer replay.
- Replays aren't in the audit log. Execution logs show a Replay details section on pages delivered by a replay. It shows the replay's Start date and, if one was set, its End date when Specific time is selected, and Earliest when Earliest retained is selected. See Monitor a subscription.
- Replay works on a range, not single messages. You can't replay one message, replay from message history, or skip ahead past a backlog.
Monitor a subscription
Topic deliveries use the platform's existing monitoring surfaces. There is no separate dashboard.
- Run console — each delivered page is a standard run.
- Error management — failed records, with retry data. Retrying reprocesses the record.
- Execution logs — from Run history > Run logs > Trace view. Account Execution log preferences apply, including whether Store payload retains page contents.
- Replay jobs — each replay runs as its own job, marked with a Replay tag in the run console and Run history.
- Used by on the topic includes this flow.
Note: Error records and retry data are independent of topic retention. A failed record — from a listener or a publish step — stays retryable from its stored retry data even after its source message ages out of the topic, or after the retention window it would have landed in has passed.
Share one listener across flows
If you reuse a single topic listener across multiple flows, those flows share one subscription and one position. Each flow sharing the export receives every delivered page.
For flows that need to move independently — different start positions, or one catching up while another doesn't — give each flow its own listener.
Listener form reference
| Section | Fields |
|---|---|
| General | Name, Description |
| Subscription | Topic, Start from (until the subscription's first run) |
| Advanced | Max wait time, Page size, Data URI template, Don't store retry data, Override trace key template |
| Custom settings | Settings |
The listener also supports the standard listener step components from Actions: transformation, output filter, and pre-save page hook.
Note: Selecting Don't store retry data doesn't affect the topic. Messages stay retained per the topic's retention period regardless.
Filter which messages your flow acts on
A topic delivers every message to every subscription. To act on only some of them, use the listener's Output filter or branch inside the flow.
Filtered messages still enter the flow before they are filtered out, so they count toward the flow's activity.