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, and how a subscription recovers its position after downtime.
Several flows can subscribe to the same topic, and each receives every message independently.
Before you begin
- 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.
- Choose the Celigo Topics application.
- Choose the step type 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.
- Name the listener and select Save.
- Enable the flow.
The flow now runs whenever messages arrive.
Note: The listener is the subscription. There is no separate subscription object to create, enable, or manage, and there is no Connection field — delivery is internal to the platform, so no connection is needed.
Choose a topic
The Topic picker lists every topic you can access, including topics not yet registered to this integration.
Select Create topic in the picker to create a topic without leaving the flow.
Saving the listener registers the selected topic to this integration automatically. See Grant access to topics.
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.
The field disappears after the first run. Once the subscription has committed a position, the setting is inert, so it is removed from the form rather than left behind as stale configuration.
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 from a specific time.
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.
Maximum messages delivered to the flow per batch. Defaults to 20 when left blank. Larger pages reduce downstream API calls; smaller pages reduce latency.
Max wait time — how long to wait for a page to fill before delivering a partial one.
Deliver a partial batch to the flow after waiting this many seconds, even if the page size hasn't been reached. Lower it for faster delivery of infrequent messages; raise it for fewer, fuller pages. Default: 5.
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. A lone message published to a topic with a 5-second max wait reaches the flow about 5 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.
Warning: Build subscribing flows to tolerate duplicates. A message can be delivered more than once — after a recovery, a retry, or an interruption — so downstream steps should produce the same result when they process the same message twice.
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
Pausing consumption is the flow's enable and disable toggle. There is no separate subscription switch.
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: A disabled flow only catches up on messages still in retention. If a flow stays disabled longer than the topic's retention period, the messages that aged out are gone. Set retention longer than your worst-case downtime. See Create and manage topics.
How a subscription decides where to resume
| 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 — another reason subscribing flows need to tolerate duplicates. A subscription is never wedged by its own position, and you are never asked to repair one manually.
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 — per run, through View execution log in the run console. Account Execution log preferences apply, including whether Store payload retains page contents.
- Used by on the topic includes this flow.
Note: Error records and retry data are independent of topic retention. A failed record stays retryable even after its source message ages out of the topic.
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 the … menu: 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.