Kafka connector

Consume Kafka topics as live keyed maps (topic_as_map → LiveMapFeed) or as raw event streams (topic_as_stream → LiveStream), and produce messages as target states — with automatic offset commits and partition rebalancing at at-least-once semantics.

Version

v 1.0.14

Last reviewed

Nov 4, 2026

The kafka connector supports Kafka as both a source (consuming messages as a live keyed map, or as a raw event stream) and a target (producing messages for declared target states).

use cocoindex::connectors::kafka::{self, KafkaMessage, KafkaProducer};

As source

The kafka connector can treat a Kafka topic as a live keyed map — each message is an upsert or delete for a key. It returns a LiveMapFeed for use with ctx.mount_each_live: a KafkaTopicMap, which is a LiveMapFeed<String, KafkaMessage>.

Setting up a consumer

Create a KafkaConsumer directly — no ContextKey needed. It is cheap to clone, and one consumer can serve several feeds.

let consumer = kafka::KafkaConsumer::connect_with_options(
    &["localhost:9092"],
    kafka::KafkaConsumerOptions {
        consumer_group: Some("my-group".to_string()),
        ..Default::default()
    },
)
.await?;

KafkaConsumer::connect(&["localhost:9092"]) connects in plaintext, without authentication and without a consumer group. KafkaConsumerOptions adds the consumer group and the connection options; SASL and TLS go through KafkaConnectOptions:

let consumer = kafka::KafkaConsumer::connect_with_options(
    &["broker:9093"],
    kafka::KafkaConsumerOptions {
        consumer_group: Some("my_app".into()),
        connect: kafka::KafkaConnectOptions {
            sasl: Some(kafka::KafkaSasl::ScramSha256 { username, password }),
            tls: Some(kafka::KafkaTls::default()),
        },
    },
)
.await?;

KafkaSasl is Plain, ScramSha256 or ScramSha512, each with a username and a password. KafkaTls::default() trusts the Mozilla root certificates; its ca_cert_pem, client_cert_pem and client_key_pem fields add a private CA and a client certificate for mutual TLS. For a connection setup these options cannot express, build an rskafka client yourself and wrap it with KafkaConsumer::from_client(client, state_id, consumer_group), where state_id is the cluster’s stable identity (for example its bootstrap-server list), never a credential.

topic_as_map()

pub fn topic_as_map(consumer: &KafkaConsumer, topic: impl Into<String>) -> KafkaTopicMap

pub fn topic_as_map_with_options(
    consumer: &KafkaConsumer,
    topic: impl Into<String>,
    options: KafkaSourceOptions,
) -> KafkaTopicMap

pub struct KafkaSourceOptions {
    /// Custom deletion predicate; defaults to "value is a tombstone (`None`)".
    pub is_deletion: Option<IsDeletionFn>,
}
pub type IsDeletionFn = Arc<dyn Fn(&KafkaMessage) -> bool + Send + Sync>;

Parameters:

• consumer — A KafkaConsumer.
• topic — The topic to consume. To consume several topics, see Consuming several topics.
• options.is_deletion — Optional predicate Fn(&KafkaMessage) -> bool for custom deletion detection on non-tombstone messages (see Deletion handling).

Returns: A KafkaTopicMap, keyed by the message key as a String (lossy UTF-8). Each item is a KafkaMessage: its topic, partition, offset, key, value, headers and timestamp fields, with key_str() and value_str() for the text forms.

Deletion handling

Messages with a None value (Kafka tombstones) are always treated as deletions. The optional is_deletion predicate provides additional deletion logic for non-tombstone messages:

// Default: only tombstones are deletions
let feed = kafka::topic_as_map(&consumer, "my-topic");

// Custom: also treat messages with a specific value as deletions
let feed = kafka::topic_as_map_with_options(
    &consumer,
    "my-topic",
    kafka::KafkaSourceOptions {
        is_deletion: Some(Arc::new(|msg: &KafkaMessage| {
            msg.value.as_deref() == Some(b"DELETED")
        })),
    },
);

Offset management

Offsets are committed automatically with at-least-once semantics. Messages are processed in parallel, but an offset is only committed after all earlier messages in the same partition have been fully processed. Messages with None keys are skipped.

A consumer with a consumer_group commits the offsets of the records a live run has processed, per partition, to the app’s state database under that group name once their child components have synced. The next live run resumes from them. A consumer without a group re-reads the topic on every run.

Readiness

The feed signals readiness after catching up to the high watermark offsets that existed when consumption started. After that, it continues consuming indefinitely until the app is stopped.

Example

use cocoindex::connectors::kafka::{self, KafkaMessage};
use cocoindex::{App, UpdateOptions};

#[tokio::main]
async fn main() -> cocoindex::Result<()> {
    let consumer = kafka::KafkaConsumer::connect_with_options(
        &["localhost:9092"],
        kafka::KafkaConsumerOptions {
            consumer_group: Some("kafka_consume".to_string()),
            ..Default::default()
        },
    )
    .await?;

    let app = App::open("kafka_consume", ".cocoindex_db").await?;

    let body = move |ctx: cocoindex::Ctx| {
        let consumer = consumer.clone();
        async move {
            let feed = kafka::topic_as_map(&consumer, "my-topic");
            ctx.mount_each_live(&"messages", feed, move |_ctx, message: KafkaMessage| {
                async move {
                    println!(
                        "message {}@{}/{}: {}",
                        message.key_str().unwrap_or_default(),
                        message.partition,
                        message.offset,
                        message.value_str().unwrap_or_default()
                    );
                    Ok(())
                }
            })
            .await
        }
    };

    let handle = app.start_update_with_options(
        UpdateOptions {
            full_reprocess: false,
            live: true,
            ..UpdateOptions::default()
        },
        body,
    )?;
    handle.result().await?;
    Ok(())
}

ctx.mount_each_live(&"messages", feed, process_fn) mounts one child component per message key. Without live: true, app.run(body) takes one snapshot of the topic up to its high watermark and stops. The full program is examples/rust/kafka_consume; it pairs with examples/rust/csv_to_kafka.

Consuming several topics

topic_as_map() keys each item by its message key alone, so one feed reads one topic. To consume several topics, create one feed per topic and mount each feed under its own mount_each_live subpath. The subpath keeps the same message key on two topics (for example, 42 on users and 42 on orders) in two separate components.

for topic in ["users", "orders"] {
    let feed = kafka::topic_as_map(&consumer, topic);
    ctx.mount_each_live(&topic, feed, move |_ctx, message: KafkaMessage| {
        async move {
            println!("{}: {}", message.topic, message.key_str().unwrap_or_default());
            Ok(())
        }
    })
    .await?;
}

The first argument of mount_each_live is the subpath; here it is the topic name. One KafkaConsumer can serve several feeds.

Each topic also writes to its own directory here, because process_message names a file by the message key only.

As a live stream

Some downstream connectors don’t model Kafka messages as a keyed map — they treat each message as an opaque event payload (e.g. the OCI Object Storage live mode ingests Object Storage event JSON delivered through OCI Streaming). For these, the kafka connector exposes a topic as a LiveStream — CocoIndex’s keyless counterpart to LiveMapFeed, an opaque sequence of messages with the same readiness and back-pressure machinery but no key/delete semantics.

Set up the consumer the same way as for topic_as_map().

topic_as_stream()

pub fn topic_as_stream(
    consumer: &KafkaConsumer,
    topics: impl IntoIterator<Item = impl Into<String>>,
) -> KafkaTopicStream

For example:

let feed = kafka::topic_as_stream(&consumer, ["users", "orders"]);
ctx.mount_each_live(&"events", feed, move |_ctx, message: KafkaMessage| {
    async move { Ok(()) }
})
.await?;

KafkaTopicStream is a LiveStream<KafkaMessage>. It is also a LiveMapFeed<String, KafkaMessage> for mount_each_live, as above: each record becomes a distinct child keyed by its "{topic}:{partition}:{offset}" position, tombstones included, and nothing is compacted or deleted. The position keys are stable, so a replayed catch-up reconciles to no-ops.

Parameters:

• consumer — A KafkaConsumer.
• topics — Topics to read, across all their partitions.

Returns: A KafkaTopicStream (a LiveStream<KafkaMessage>) that delivers every record, tombstones included, to its subscriber. Offset management and readiness behave identically to topic_as_map().

KafkaTopicStream::payloads()

pub fn payloads(self) -> KafkaPayloadStream // a LiveStream<Vec<u8>>

A view over the same stream that yields each message’s value (the byte payload) instead of the full message. Null-valued messages (Kafka tombstones) are filtered out of this view; consumers that need tombstone semantics should subscribe to the KafkaTopicStream directly.

This is the typical input for sources that consume opaque event payloads:

let events = kafka::topic_as_stream(&consumer, ["object-storage-events"]).payloads();
let walker = oci_object_storage::list_objects_live(
    &client,
    "my-namespace",
    "my-bucket",
    oci_object_storage::ListOptions::default(),
    events,
);

The OCI live view acknowledges each event to the stream once the object it names has synced, so the consumer group’s offset never passes an unprocessed event.

For a complete app using this pattern, see the OCI Object Storage live-mode example.

As target

The kafka connector provides target state APIs for producing messages to Kafka topics. Topics are user-managed (CocoIndex does not create or drop topics) — CocoIndex only produces messages to them.

Setting up a producer

use cocoindex::connectors::kafka::{self, KafkaProducer};
use cocoindex::Environment;

cocoindex::context_key!(static PRODUCER: KafkaProducer);

let producer = KafkaProducer::connect(&["localhost:9092"]).await?;
// The topic is user-managed; create it up front if absent (idempotent).
producer.ensure_topic(&topic, 1).await?;

// The producer is provided under a `ContextKey`; the target names the key.
let app = Environment::builder()
    .provide_key(&PRODUCER, producer)
    .build()
    .await?
    .app("CsvToKafkaRust")
    .await?;

KafkaProducer::connect_with_options(servers, KafkaConnectOptions { sasl, tls }) adds SASL and TLS, as for the consumer, and KafkaProducer::from_client(client, state_id) wraps an rskafka client you built. ensure_topic(topic, num_partitions) is an explicit helper outside reconciliation; CocoIndex never creates or drops a topic on its own.

Declaring target states

Topics (parent state)

Declares a topic as a target state. Returns a KafkaTopicTarget for declaring messages. kafka_topic_target() takes the same arguments and builds the topic’s target state alone, for mount_target or declare_target_state_with_child; declare_kafka_topic_target() and mount_kafka_topic_target() wrap it.

pub fn declare_kafka_topic_target(
    ctx: &Ctx,
    producer: &ContextKey<KafkaProducer>,
    topic: impl Into<String>,
    options: KafkaTopicOptions,
) -> Result<KafkaTopicTarget>

pub async fn mount_kafka_topic_target(
    ctx: &Ctx,
    producer: &ContextKey<KafkaProducer>,
    topic: impl Into<String>,
    options: KafkaTopicOptions,
) -> Result<KafkaTopicTarget>

pub fn kafka_topic_target(
    ctx: &Ctx,
    producer: &ContextKey<KafkaProducer>,
    topic: impl Into<String>,
    options: KafkaTopicOptions,
) -> Result<TargetState<TopicSpec>>

pub struct KafkaTopicOptions {
    pub deletion_value_fn: Option<DeletionValueFn>,
}
/// Takes the key as Kafka stores it: a `&str` key is its UTF-8 bytes.
pub type DeletionValueFn = Arc<dyn Fn(&[u8]) -> Vec<u8> + Send + Sync>;

Parameters:

• producer — A ContextKey<KafkaProducer> identifying the producer to use.
• topic — The Kafka topic name; it must be non-empty.
• options.deletion_value_fn — Optional callback that produces a deletion value for a given key (see Deletion handling).

Returns: A KafkaTopicTarget. declare_kafka_topic_target declares the topic in the current component, and its messages resolve at the component’s commit; mount_kafka_topic_target mounts it in the foreground, so you can declare messages at once:

let target = kafka::mount_kafka_topic_target(
    &ctx,
    &PRODUCER,
    "my-topic",
    kafka::KafkaTopicOptions::default(),
)
.await?;

Messages (child states)

Once a KafkaTopicTarget is resolved, declare target states to produce messages:

pub fn KafkaTopicTarget::declare_message(
    &self,
    ctx: &Ctx,
    key: &str,
    value: impl AsRef<[u8]>,
) -> Result<()>

pub fn KafkaTopicTarget::declare_message_bytes(
    &self,
    ctx: &Ctx,
    key: &[u8],
    value: impl AsRef<[u8]>,
) -> Result<()>

Parameters:

• key — The message key, used as the stable identity for change detection; it must be non-empty. declare_message takes a &str and sends its UTF-8 bytes; declare_message_bytes takes a binary key. A &str key and its UTF-8 bytes are the same Kafka key: declare each key in one form only. If a key changes form between runs (for example, from "a" to b"a"), CocoIndex publishes the new form and sends no deletion message for the old form, so the record stays.
• value — The message value: any byte slice, so a &str, String or Vec<u8> works.

CocoIndex fingerprints the value and only produces a message when it has changed since the last run.

Deletion handling

While the topic is still declared, a message target state that is no longer declared gets a deletion message. The behavior depends on deletion_value_fn:

• Without callback (default): Produces a message with the key and no value (Kafka tombstone).
• With callback: Calls deletion_value_fn(key) to produce the deletion value.

// Tombstone on deletion (default)
let target = kafka::mount_kafka_topic_target(
    &ctx,
    &PRODUCER,
    "my-topic",
    kafka::KafkaTopicOptions::default(),
)
.await?;

// Custom deletion value
let options = kafka::KafkaTopicOptions {
    deletion_value_fn: Some(Arc::new(|key: &[u8]| {
        format!("deleted:{}", String::from_utf8_lossy(key)).into_bytes()
    })),
};
let target = kafka::mount_kafka_topic_target(&ctx, &PRODUCER, "my-topic", options).await?;

Abandoning a topic

Removing the topic declaration itself — or calling App::drop_state — abandons the topic: CocoIndex stops producing to it and emits no tombstones for the messages it produced earlier. Renaming the topic in code abandons the old topic the same way. Avoid reusing an abandoned topic in a later declaration: its stale keys remain and are never tombstoned, and keys you declare again are republished as new even when their values are unchanged.

To tombstone everything before abandoning a topic, drain it first:

1. Keep the topic declared, but declare no messages in it.
2. Run an update — every previously produced key gets a deletion message.
3. Remove the topic declaration.

Example

use cocoindex::connectors::kafka::{self, KafkaProducer};
use cocoindex::prelude::*;
use cocoindex::Environment;

cocoindex::context_key!(static PRODUCER: KafkaProducer);

/// Parse one CSV file and declare one message per row. Memoized on the file:
/// an unchanged file keeps its messages without re-parsing. The topic target
/// is one per app, so it is not part of the memo key.
#[cocoindex::function(memo, memo_key(target = skip))]
async fn process_csv(ctx: &Ctx, file: &FileEntry, target: &kafka::KafkaTopicTarget) -> Result<()> {
    let text = file.read_text().await?;
    // ... for each CSV row, build its message key and JSON value, then:
    // target.declare_message(ctx, key, value)?;
    Ok(())
}

async fn index(producer: KafkaProducer, topic: String) -> Result<()> {
    // The topic is user-managed; create it up front if absent (idempotent).
    producer.ensure_topic(&topic, 1).await?;

    let app = Environment::builder()
        .db_path(std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(".cocoindex_db"))
        .provide_key(&PRODUCER, producer)
        .build()
        .await?
        .app("CsvToKafkaRust")
        .await?;

    let stats = app
        .run(move |ctx| {
            let topic = topic.clone();
            async move {
                let target = kafka::mount_kafka_topic_target(
                    &ctx,
                    &PRODUCER,
                    topic,
                    kafka::KafkaTopicOptions::default(),
                )
                .await?;

                let files = walk_dir("./data")
                    .path_matcher(PatternFilePathMatcher::include(["**/*.csv"])?)
                    .live();
                // One processing component per file: each file's messages sync
                // and clean up with that file.
                ctx.mount_each_live(&"process_csv", files, move |ctx, file: FileEntry| {
                    let target = target.clone();
                    async move { process_csv(&ctx, &file, &target).await }
                })
                .await
            }
        })
        .await?;
    println!("{stats}");
    Ok(())
}

The full program, with the CSV parsing, is examples/rust/csv_to_kafka. KafkaTopicTarget is not a memo input, so a memoized function that takes it skips it in its memo key with memo_key(target = skip).