Ingest a Kafka Message Stream Programmatically

Use this quickstart when your data arrives as messages on a Kafka topic and you want to script the subscription. To build the same pipeline on the Data Fusion canvas, the visual pipeline editor in C3 AI Studio, see Connect with Streaming Source System. A topic is a named channel on a message broker, the service that receives messages from producers and delivers them to consumers.

You model the data, create the source system and source collection, connect credentials and start the stream, send test messages, verify the stored entities, and stop the stream. A later section shows how to make the pipeline part of your package. "Fix common problems" covers the errors you are most likely to see.

A streaming source has no sync or process call. After you start the source collection, each message runs through the transform and persists to the target entity Type as it arrives on the topic. The worked example consumes JSON order messages from a Confluent Cloud topic. The same pattern applies to Amazon Kinesis, Azure Event Hubs, and Google Cloud Pub/Sub through their credential Types.

You work in two places. You define Types and the transform in your IDE as package source, which the C3 Agentic AI Platform syncs to the application. You run runtime commands in the C3 AI Console of your application.

A streaming pipeline links these Types:

• CloudMessageSourceSystem: the source system, which is the connection to the message broker.
• ApacheCredentials: the broker endpoint and API key credentials.
• CloudMessageSourceCollection: the source collection, which is the subscription that binds one topic to a Source Type.
• Source: the Source Type, which describes one message.
• Transform: the transform, which maps each message to the target.
• A target entity Type: the Type that stores the data in the application.

Copy link to this section

Prerequisites

Before you start, make sure you have the following:

• A C3 Agentic AI Platform application in dev mode, with access to the C3 AI Console. The platform accepts new Types, transforms, source systems, and source collections only while the application is in dev mode.
• A role that allows data integration from code. To choose one, see Data Integration Roles and Permissions.
• A Kafka broker endpoint, an API key and secret that can read the topic and the consumer group $Default, and an existing topic. The example uses a Confluent Cloud topic named Orders. The other example names (confluentKafka, ordersStream, SourceOrdersJSON, SourceOrdersAddress, OrderReading, and SourceOrdersOutboundQueue) stand in for your own source system, collection, and Type names. Replace the example names, and the topic name, with your own.
• The package source for your application, open in your IDE, such as VS Code with the C3 AI extension.

Copy link to this section

Step 1: Model the data

Define what a message looks like, what the application stores, and how one maps to the other. In your IDE, add these files to the package source, then sync the package to the application.

Define the Source Type that models one JSON message. A nested JSON object maps to its own Type. Put each Type in its own .c3typ file, because the platform cannot parse a file that holds two type declarations.

type SourceOrdersJSON mixes Source {
  ordertime: int
  orderid: int
  itemid: string
  orderunits: double
  address: SourceOrdersAddress
}

type SourceOrdersAddress {
  city: string
  state: string
  zipcode: int
}

Define the target entity Type:

entity type OrderReading {
  orderTime: int
  item: string
  units: double
  city: string
  state: string
}

Define the transform. The projection lists each target field and the expression that fills it from the message. Nested message fields, such as address.city, project with dot notation. Save the transform as metadata/Transform/SourceOrdersJSON-OrderReading.js:

data = {
  name: 'SourceOrdersJSON-OrderReading',
  source: 'SourceOrdersJSON',
  target: 'OrderReading',
  projection: {
    id: concat('order_', orderid),
    orderTime: ordertime,
    item: itemid,
    units: orderunits,
    city: address.city,
    state: address.state,
  },
};

Optionally, define a test producer, a Type that sends messages to the topic. A streaming pipeline shows nothing until a message arrives. This one-line Type lets you send test messages from the Console in Step 4, before real producers exist:

type SourceOrdersOutboundQueue mixes CloudOutboundMessage<SourceOrdersJSON>

Save the files and wait for the platform to finish syncing the package source before you continue.

Copy link to this section

Step 2: Create the source system and collection

In the Console, create the source system, which represents the broker, and the source collection, which is the subscription. The source collection names the topic, the message format, and the Source Type that models one message. Create both before you set credentials in Step 3: setCredentials stores the credentials for the topics of the source collections that exist when it runs.

CloudMessageSourceSystem.make({ name: 'confluentKafka' }).upsert();

CloudMessageSourceCollection.make({
  name: 'ordersStream',
  sourceSystem: CloudMessageSourceSystem.forName('confluentKafka'),
  source: SourceOrdersJSON,
  external: true,
  cloudMessageBrokerName: 'Orders',
  cloudMessageBrokerType: 'ApacheKafkaTopic',
  batchSize: 10,
  timeoutMillis: 5000
}).upsert();

var config = CloudMessageSourceCollection.forName('ordersStream').config();
config.setConfigValue('contentTypeOverride', 'application/json', ConfigOverride.APP);
config.setConfigValue('doNotChunk', true, ConfigOverride.APP);
config.setConfigValue('doNotArchiveStatus', false, ConfigOverride.APP);
config.setConfigValue('doNotArchiveSources', true, ConfigOverride.APP);
config.setConfigValue('doNotArchiveFailedSources', false, ConfigOverride.APP);
config.setConfigValue('processMode', 'ON_ARRIVAL', ConfigOverride.APP);

The source collection fields:

• cloudMessageBrokerName: The topic name on the broker, Orders in this example. Use the name of your topic here, in the Step 4 producer code, and in the package metadata file.
• cloudMessageBrokerType: The broker Type as a string. Use ApacheKafkaTopic for Kafka, AwsKinesis for Amazon Kinesis, or AzureEventHub for Azure Event Hubs.
• external: true: The topic is managed outside the platform, so the platform does not try to create or delete it.
• batchSize and timeoutMillis: The batch settings for consuming messages from the topic. timeoutMillis is in milliseconds.

The configuration settings, set with setConfigValue on the source collection's configuration:

• contentTypeOverride: Sets the media type of the message content.
• doNotChunk: Keeps each message whole instead of splitting it into chunks.
• doNotArchiveStatus: Set to false, archives each SourceStatus record, the run status the platform writes.
• doNotArchiveSources: Set to true, skips archiving the processed message content.
• doNotArchiveFailedSources: Set to false, writes messages that fail to parse to the failed-sources location.
• processMode: Set to ON_ARRIVAL, processes messages as they arrive. Set the process mode before you start the source collection. A registered source collection left in MANUAL mode, the default, skips the messages that arrive. The platform reads the skipped messages but loads none of them, and they do not load later. To read the current mode, run CloudMessageSourceCollection.forName('ordersStream').config().processMode.

The Console route suits iteration in a dev application. To deploy the pipeline with your package instead, follow "Make the pipeline part of your package".

Copy link to this section

Step 3: Connect credentials and start the stream

In the Console, set the credentials on the source system and start the subscription. setCredentials stores the credentials in your application. The consumer uses the stored credentials to reach the broker. startOnArrivalMode subscribes the source collection to the topic; from then on, ingestion is continuous and needs no further calls.

var apacheCreds = ApacheCredentials.make({
  endpoint: '<broker-endpoint>',
  apacheApiKeyCredentials: { apiKey: '<api-key>', apiSecret: '<api-secret>' }
});
CloudMessageSourceSystem.forName('confluentKafka').setCredentials(apacheCreds);

CloudMessageSourceCollection.forName('ordersStream').startOnArrivalMode();

Replace the following placeholders with your values:

• <broker-endpoint>: The broker bootstrap endpoint, for example broker.example.com:9092.
• <api-key> and <api-secret>: The broker API key credentials.

To confirm the connection, run CloudMessageSourceCollection.forName('ordersStream').ping(). The ping call returns {"reachable":true} when the platform can reach the broker.

The pipeline does not use broker administration calls such as ApacheKafkaTopic.configKey, fromResourceName, and receiveMessages. Broker administration calls require cluster administrator access.

Copy link to this section

Step 4: Send test messages

If you added the optional SourceOrdersOutboundQueue Type in Step 1, use it in the Console before real producers exist. Sending test messages checks the broker, the subscription, the transform, and persistence. The code points the producer at the topic, builds ten sample messages, and sends them to partition 0. The ordertime value is a timestamp in epoch milliseconds.

SourceOrdersOutboundQueue.setBrokerName('Orders', 'ApacheKafkaTopic');
SourceOrdersOutboundQueue.setBatchSize(10);

var cAry = C3.ArrayBuilder.fromJson({ type: 'ArrayBuilder<SourceOrdersJSON>' });
for (var i = 1; i <= 10; i++) {
  cAry.push(SourceOrdersJSON.make({
    ordertime: 1751500000000 + i * 3600000,
    orderid: 20260700 + i,
    itemid: 'Item_' + i,
    orderunits: 1.5 + i,
    address: SourceOrdersAddress.make({ city: 'City_' + i, state: 'State_' + (i % 3), zipcode: 94000 + i })
  }));
}
SourceOrdersOutboundQueue.sendBatch(cAry.build(), '0');

The second argument to sendBatch is the partition, as a string. The call returns an empty map, {}, when every message sends cleanly.

Delivery is at-least-once, not exactly-once, so a message can process twice. The transform builds a deterministic id from orderid, so the same message always produces the same ID. The platform upserts a redelivered message onto the same entity instead of creating a duplicate.

Copy link to this section

Step 5: Verify the data

In the Console, fetch the target entities and the status records to confirm that the messages persisted. With on-arrival mode there is nothing to trigger.

c3Grid(OrderReading.fetch({ order: 'ascending(id)' }));

SourceStatus.fetch({
  filter: "sourceCollectionName=='ordersStream'",
  order: 'descending(meta.created)'
});

c3Grid shows the result as a table in the Console. The grid shows one OrderReading entity per message, with IDs order_20260701 through order_20260710, populated through the transform, including the nested address fields. Each consumed batch of messages produces one or more SourceStatus records with a completed state. New messages sent to the topic keep appearing as entities without any further action.

If no rows appear, check the SourceStatus records and the table in "Fix common problems".

Copy link to this section

Step 6: Stop the stream

The subscription keeps consuming messages until you stop it. When you finish testing, stop the source collection in the Console:

CloudMessageSourceCollection.forName('ordersStream').stopOnArrivalMode();

stopOnArrivalMode unsubscribes the source collection from the topic. stopOnArrivalMode does not reset processMode, which stays ON_ARRIVAL, so startOnArrivalMode resumes the stream later. If another source collection on the same topic is in ON_ARRIVAL mode, even one that is stopped, the call leaves the subscription in place, and startOnArrivalMode on this collection does nothing. Set the other collection's process mode to MANUAL first.

Copy link to this section

Make the pipeline part of your package

Declare the source system and source collection as package metadata when the pipeline must deploy with your package. Package metadata deploys with the package, so the source collection already exists when you run setCredentials.

The folder under metadata/ or config/ must match the Type name. In metadata, refer to the Source Type by name and to the source system by Type and name.

1. Create metadata/CloudMessageSourceSystem/confluentKafka.json:

   JSON

   Copy code to clipboard

   { "name": "confluentKafka" }

2. Create metadata/CloudMessageSourceCollection/ordersStream.json:

   JSON

   Copy code to clipboard

   {
     "name": "ordersStream",
     "source": "SourceOrdersJSON",
     "sourceSystem": { "type": "CloudMessageSourceSystem", "name": "confluentKafka" },
     "external": true,
     "cloudMessageBrokerName": "Orders",
     "cloudMessageBrokerType": "ApacheKafkaTopic",
     "batchSize": 10,
     "timeoutMillis": 5000
   }

3. Create config/CloudMessageSourceCollection.Config/ordersStream.json with the configuration settings from Step 2:

   JSON

   Copy code to clipboard

   {
     "name": "ordersStream",
     "contentTypeOverride": "application/json",
     "doNotChunk": true,
     "doNotArchiveStatus": false,
     "doNotArchiveSources": true,
     "doNotArchiveFailedSources": false,
     "processMode": "ON_ARRIVAL"
   }

4. Sync the package to the application.

Keep credentials out of the package. Credentials are secrets: set them at runtime with setCredentials, as in Step 3.

Copy link to this section

Fix common problems

Find the symptom you see, then apply the fix in the same row.

Copy link to this section

See also

• Integrate Data Programmatically
• Connect with Streaming Source System
• Authenticate Streaming Sources
• Monitor Your Data Pipeline
• Connect C3 AI Application to Apache Kafka
• Configure Streaming Sources
• Ingest Files from Amazon S3 Programmatically
• Ingest a Snowflake Table Programmatically
• Query a Snowflake Table Programmatically with Virtualization
• Combine Multiple Sources with a Canonical