configuring-topics

<!-- TIER:1 -->

Configuring Topics (Event Streams)

A topic is a named, durable message channel. Publishers append messages; every listener receives all of them, reading from its own position at its own pace; and a message stays readable until the topic's retention window ends, whether or not anything read it. A topic decouples the flow that produces an event from the flows that consume it: add a consumer without touching the producer, pause one consumer while the others carry on, and let it catch up from where it stopped.

A topic is an account resource that lives in one environment, outside any integration -- like a lookup cache. Flows use it through two steps, neither of which takes a connection:

• Publisher -- an import with adaptorType: "TopicImport". Every record it receives becomes one message.
• Listener -- an export with adaptorType: "TopicExport" and type: "stream", the page generator of a flow that runs when messages arrive.

Concerns when working with topics:

• The topic -- name, retention, the license that gates it, and registration on the integrations whose flows use it
• Publishing -- a publisher step in a flow, API, or Tool, or a direct publish call
• Listening -- a listener step, and the flow-enable transition that subscribes it
• Messages -- reading what is retained (never moves a listener) and purging it (irreversible)
• Positions -- each listener's offset on the topic: where its next run resumes

How to Run the Steps

This skill names each step by what it reads or changes; the exact calls depend on how you reach the account. Read the one reference that matches your tools -- both follow the step numbers of How to Build an Event Stream below, then the headings of Messages and Positions, Diagnosing a Listener That Never Runs, and Delete, Restore, and Copies:

• Celigo MCP tools (upsert_export, upsert_import, upsert_flow, patch_flow, ...) -- references/mcp.md. The deployed server has no topic tools: it writes the listener and publisher steps and their flows, while creating, registering, publishing to, reading, purging, and deleting topics take the CLI, the Celigo UI, or the REST API.
• The celigo CLI -- references/cli.md

Listing topics and reading messages, positions, dependencies, and the audit log only read. Creating, updating, registering, deregistering, and deleting a topic write to the account. Publishing reaches every flow subscribed to the topic -- each one processes the message -- and purging deletes every retained message for good, so confirm both with the user first.

When to Use a Topic

Need	Use
One event, several independent consumers	A topic -- each listener flow receives every message
Hand records from one flow to others	A topic -- chaining with _runNextFlowIds passes no records (building-flows > Configure chaining)
A consumer that can be paused and catch up	A topic -- a listener whose flow is disabled keeps its position and resumes from it, as long as the messages are still retained
Enrich records mid-flow from published data	Not a topic -- a listener can't be a lookup step. Listen in its own flow, or keep the reference data in a lookup cache
React to events in an outside system	A listener on that system (webhook, NetSuite or Salesforce listener, CDC) -- configuring-exports > Listeners. An outside system can also publish to a topic directly, with an API token scoped to it

Quick Reference

Topic Fields

Field	Rules
name	Send one: the API saves a topic without a name, and it can't be told apart in lists. Up to 200 characters. Names aren't unique -- steps bind the _id, so a rename never breaks one
description	Up to 5,120 characters; display only
retentionMs	Milliseconds a message stays readable, read or not. Default 604800000 (7 days); at least 3600000 (1 hour); at most 7776000000 (90 days) and the license's topics.retentionCeilingDays. Shortening it drops messages older than the new window
externalId	Your own key, to find the topic again without storing its id (a list filter)
draft	Set only on create: the topic deletes itself 48 hours later unless an update sets draft to false
__integrationIds	Create only: registers the new topic on these integrations. Never stored or echoed back; a registration that fails is reported in __failedIntegrationRegistrations, and the topic is still created

Only name and description can be changed on their own. Any other change is a full update, and a full update clears description and externalId and resets retentionMs to 7 days when the body leaves them out -- always send the complete topic.

License and Quota

Event Streams is licensed. The account license carries topics.enabled, topics.topicsAllowed (how many topics the account may hold), and topics.retentionCeilingDays. Listing topics works on every account; creating or updating one, and every message operation, needs the entitlement -- without it they fail with topics_subscription_required (422 on topic writes, 403 on message calls). The quota is checked when a topic is created or restored from the recycle bin (topics_limit_exceeded), nothing warns as the account approaches it, and topics in the recycle bin don't count.

The Two Steps

Step	Resource	Required	Defaults	Schema
Publisher	import, adaptorType: "TopicImport"	topic._topicId	topic.batchSize 1000	configuring-imports topic.yml
Listener	export, adaptorType: "TopicExport", type: "stream"	topic._topicId	topic.startPosition latest, topic.maxWaitTimeMs 30000	configuring-exports topic.yml

topic._topicId is the topic's _id, never its name. Neither step takes a _connectionId: with one, the save validates the step as that connection's adaptor and fails. The step rules live in configuring-exports > Listeners and configuring-imports > Topic Imports; this skill covers the topic they share.

Schema Index

The listener and publisher step schemas are the topic.yml files linked under The Two Steps. The topic's own request and response shapes -- the topic, publish and its per-message results, a page of messages, a listener's position, and the purge result -- are on the API reference's Topics page (agent-readable markdown).

Related Skills

• configuring-exports > Listeners -- the topic listener step (TopicExport)
• configuring-imports > Topic Imports -- the publisher step (TopicImport)
• building-flows > Create disabled, test, enable -- enabling a flow is what subscribes its listener
• configuring-imports > Writing Exactly Once -- idempotent writes in listener flows
• managing-integrations -- registration on integrations, cloning, and promotion across environments
• managing-api-tokens > Decide the scope -- tokens that can publish to and read only named topics
• configuring-lookup-caches -- the other environment resource registered on integrations; use it for enrichment

<!-- TIER:2 -->

How to Build an Event Stream

1. Plan publishers and listeners

• The message. What one message stands for (an order, an order line), and its shape: any JSON object up to 1 MiB serialized. A listener's flow receives each message as its record, exactly as it was published -- no envelope, no timestamp.
• Who publishes. A publisher step in a flow, an API, or a Tool, or a direct publish call -- from outside Celigo too, with an API token scoped to the topic.
• Who listens. One listener flow per consumer. Every listener receives every message, from its own position.
• Retention. At least the longest time a listener may be paused or failing before someone fixes it. Messages older than the window are dropped whether or not anything read them; a listener paused longer than that never sees them and resumes from the earliest message left.
• Duplicates. The same message can be processed more than once (step 7), so plan an idempotent write for every listener flow.

2. Check for an existing topic

List the account's topics, filter by externalId, or list what one integration has registered. Names aren't unique: before reusing a topic, confirm it by _id and description, and see what already publishes to and listens on it (its dependencies).

3. Create the topic and register it

Create the topic with a name, a description of what it carries, and a retention period; add an externalId if something must find it again by key. Flows publish and listen on topics registered on their integration. Register the topic on create (__integrationIds) or afterwards through the integration's topic registration -- or let the flow do it: saving a flow with a step bound to a topic registers that topic on the flow's integration. Registration adds no alias, unlike a lookup cache's.

4. Build the publisher

An import with adaptorType: "TopicImport" and topic._topicId, with no connection and no create, update, or upsert logic: every record becomes one message.

• Without a mapping, the message is the record as the step received it; a mapping reshapes it.
• With oneToMany: true and pathToMany, each element of that array is published as its own message, carrying the rest of the parent record (without the array) under _PARENT.
• topic.batchSize (default 1000) is how many records go per publish request; lower it when messages are large.

The step can publish from a flow, from an API, from a Tool reached through a ToolImport step, or on its own through an import invoke. A flow or API test run publishes nothing -- import steps are mocked there -- while an invoke publishes for real to every subscribed flow.

5. Build the listener and its flow

An export with adaptorType: "TopicExport", type: "stream", and topic._topicId, with no connection, no cdc object, and no isLookup. Make it the flow's page generator and create the flow with disabled: true and no schedule.

• topic.startPosition -- latest (the default) delivers only messages published after the listener subscribes; earliest first replays every message the topic still holds, the moment it subscribes. It's read when the listener is created and whenever its _topicId changes, so to reprocess retained history create a new listener with earliest.
• topic.maxWaitTimeMs -- how long the listener collects messages before handing the flow a batch, at least 1000 (default 30000). The first run after a publish starts roughly that long after it; lower it for latency on a quiet topic. The export's pageSize caps the records per batch.

Without type: "stream" the export still saves but runs as a batch export that never starts on a publish -- see configuring-exports > Listeners for the repair.

6. Enable the flow to subscribe

The listener subscribes when its flow goes from disabled to enabled; within about half a minute its position appears (step 3 of Diagnosing a Listener That Never Runs) -- or, on a topic nothing has been published to yet, with the first publish, which the listener still receives. A flow created already enabled never subscribes: it doesn't run when messages are published, and its listener has no position until the flow is disabled and enabled again. Under latest, anything published before the subscription formed is skipped -- enable first, then publish. After repairing a listener, disable and re-enable its flow.

Then publish a test message (or invoke the publisher with one record) and confirm the flow ran.

7. Make every listener flow's writes idempotent

A message can reach the destination more than once, so key each listener flow's write on the message's business key (configuring-imports > Writing Exactly Once):

• An earliest listener replays every retained message when it subscribes.
• Every listener processes every message -- including a listener in a copy of the flow cloned within the same environment, which keeps the topic.
• A retried error re-sends its record (building-flows > Duplicate Records).
• A listener whose saved position expired or was purged resumes from the earliest message still retained.

Messages and Positions

Publish. The body is {"messages": [ {...}, ... ]} -- each message one JSON object, at most 1 MiB (1,048,576 bytes) serialized; a bare array fails with 400 invalid_messages, and an empty array publishes nothing. A 200 means the body was valid, not that every message was published: read each responses[].statusCode. An oversized message fails on its own with 422 topic_publish_event_too_large while the rest of the batch publishes. The first publish to a new topic provisions its stream and takes a few seconds longer than later ones.

Read. Messages come back oldest first, at most 100 per page (default 50), each as {timestamp, payload}, where payload is the message as published. Follow nextPageURL until it's absent, and narrow by publish time with time_gt and time_lte (epoch milliseconds). There's no search. Reading never moves any listener's position.

Purge. Deletes every retained message and keeps the topic, its retention, its registrations, and its subscriptions; it can't be undone, and the topic's audit log records it as a purge of messages. Offsets keep counting after a purge, and enabled listeners receive the next messages as usual. Purge empties the topic; deleting it removes the topic -- choose as with a lookup cache.

Positions. For one listener on one topic the position is {hasCursor, offset}, where offset is the next message the listener will read, counted from the start of the stream.

• hasCursor: false (offset -1) -- most often the listener has never subscribed (its flow hasn't gone from disabled to enabled since it was saved, it was saved without type: "stream", or it doesn't listen to this topic), or nothing has ever been published to the topic: an enabled listener's position appears with the first publish, and it still receives that message.
• A disabled flow keeps its listener's position, and on re-enable the flow receives everything published meanwhile.
• While no message has expired or been purged, the retained message count minus offset is the listener's backlog.
• When the message at a listener's position is no longer retained -- it expired or was purged -- the listener still reports hasCursor: true with that offset, and its next run resumes from the earliest message still retained, whatever its startPosition.

Diagnosing a Listener That Never Runs

1. Read the listener export. It needs adaptorType: "TopicExport", type: "stream", and the _id of the topic the publisher uses in topic._topicId. Without the type, repair the export as in step 5, then toggle the flow.
2. Read the flow. It must be enabled. A flow created enabled has never subscribed: disable it, then enable it.
3. Read the listener's position. hasCursor: false most often means it never subscribed (steps 1 and 2), or that the topic has never received a message (step 4) -- then the position appears with the first publish. hasCursor: true means it's subscribed.
4. Read the topic's messages. While nothing has expired or been purged, the retained count minus offset is what the listener hasn't read yet. Nothing unread, or no message with a timestamp after the flow was enabled, means nothing was published since it subscribed: check the publisher. Flow and API test runs publish nothing, and a publish answers 200 even when a message was rejected.
5. Allow for the batch window. A run starts about maxWaitTimeMs (default 30 seconds) after a publish. Then read the flow's runs and errors as for any flow (troubleshooting-flows).
6. Records missing from runs. Under latest, messages published before the subscription formed were skipped; messages that expired past retentionMs (or were purged) before the listener read them are gone, and it carried on from the earliest message left. To reprocess what is still retained, add a new listener with startPosition: "earliest".

Delete, Restore, and Copies

• Delete is refused while anything binds the topic: any export or import with its topic._topicId -- even one in no flow -- or a flow that contains one. The 422 dependencies_not_deleted names each blocker; read the topic's dependencies, then delete or repoint each one.
• Deregistering is refused while a step in that integration binds the topic (400 not_allowed). Deleting the topic needs no deregister first: the delete drops it from every integration it was registered on.
• A deleted topic sits in the recycle bin for 30 days, missing from lists and answering 404 on reads. Restoring it brings back its retained messages but not its registrations -- register it again. Purging it from the bin deletes it and its messages for good, and steps that bound it must be purged from the bin first.
• A clone within the same environment keeps the topic. A cloned export, import, or whole flow keeps topic._topicId, so the clone of a listener flow is one more consumer once enabled. A copy in another environment needs that environment's topic: after cloning into another environment, installing a template, or pulling changes, check each topic step's topic._topicId.

<!-- TIER:3 -->

Pre-Submit Checklist

• The license has Event Streams enabled, and a create leaves the account within topicsAllowed
• retentionMs is between one hour and 90 days and within the license's retentionCeilingDays
• A full update of the topic carries description, externalId, and retentionMs
• The topic is registered on every integration whose flows publish or listen (or a flow save registers it)
• Listener: adaptorType: "TopicExport", type: "stream", topic._topicId set to the topic's _id; no _connectionId, cdc, or isLookup
• Publisher: adaptorType: "TopicImport", topic._topicId set to the topic's _id; no _connectionId
• startPosition chosen deliberately -- earliest replays everything retained
• The listener flow was created with disabled: true and no schedule, and is enabled only when ready
• Every listener flow's write is idempotent
• Every publish result was checked per message

Gotchas

1. A listener flow created enabled never subscribes. It doesn't run on a publish and its listener has no position until the flow goes from disabled to enabled.
2. latest skips what came before the subscription. Messages published before the flow was enabled -- or while a flow created enabled sat unsubscribed -- never reach that listener.
3. earliest replays everything retained the moment the listener subscribes, without any new publish. Make sure the flow's writes are idempotent first.
4. A listener saved without type: "stream" saves without error and never runs on a publish; a PATCH can't add the type.
5. A full update erases what it omits. description and externalId are cleared and retentionMs goes back to 7 days.
6. A publish 200 isn't per-message success. Read every responses[].statusCode; a message over 1 MiB fails alone.
7. Test runs publish nothing; an invoke publishes for real, to every subscribed flow.
8. Purge is irreversible. It keeps subscriptions and offsets keep counting -- there's no rewind.
9. Names aren't unique. Bind and look up by _id (or externalId).
10. Any binding blocks the delete, even an export or import that's in no flow -- and once the topic is in the recycle bin, the deleted steps that bound it block its purge until they're purged.
11. Restore doesn't re-register. Register the restored topic on each integration again.
12. A cloned listener flow is another consumer. Same-environment clones keep topic._topicId.

Common Errors

Error	Cause	Fix
topics_subscription_required (422 on create or update, 403 on message calls)	The license doesn't include Event Streams	Check the license's topics.enabled; the account owner contacts Celigo
422 topics_limit_exceeded	The account holds topicsAllowed topics	Delete or reuse an unused topic (topics in the recycle bin don't count), or have the account owner ask Celigo to raise the limit
422 min on retentionMs	Under one hour (3600000)	Use at least 3600000
422 topics_retention_exceeds_entitlement and max	Over the license ceiling or 90 days	Stay within retentionCeilingDays and 7776000000
422 ... is not a whitelisted property on a PATCH	Only /name and /description patch	Send a full update with every field
400 invalid_messages on publish	The body isn't {"messages": [...]}	Wrap the messages in a messages array
topic_publish_event_too_large in responses[]	One message over 1 MiB serialized	Trim or split that message; the rest were published
422 invalid_ref "... is not a valid topic._topicId"	Not a topic _id in this environment -- often the name	Look the id up by name or externalId
422 missing_required_field "http subschema not defined"	A _connectionId on a topic step	Remove the connection
422 not_supported "Lookup functionality is not available for Stream Export"	isLookup: true on a listener	Listen in a separate flow; enrich from a lookup cache
422 dependencies_not_deleted on delete	An export, import, or flow still binds the topic	Delete or repoint each one named, then retry
400 not_allowed on deregister	A step in the integration binds the topic, or it isn't registered there	Repoint or delete the steps first
404 "already in the Recycle Bin"	The topic was deleted	Restore it, then register it again
Listener flow never runs	Not subscribed, nothing published, or the batch window	Diagnosing a Listener That Never Runs