kafka

name	kafka
summary	Integrate Couchbase with Apache Kafka using the Kafka Connect Couchbase connector
description	Integrate Couchbase with Apache Kafka using the Kafka Connect Couchbase connector
allowed-tools	Bash
compatibility	Couchbase Server >=6.0, Kafka Connector >=4.0
metadata	{"last_verified":"2026-05","min_server_version":"6.0","handoff":[{"condition":"user asks about Eventing or document triggers","skill":"eventing"},{"condition":"user asks about XDCR replication","skill":"xdcr"}]}

Couchbase Kafka Connector

Kafka Connect plugin that uses DCP (Database Change Protocol) to stream every mutation, deletion, and expiration from Couchbase to Kafka (Source) or write Kafka records to Couchbase (Sink).

Installation

# Confluent Hub
confluent-hub install couchbase/kafka-connect-couchbase:4.2.6

# Or add ZIP to plugin.path in connect-*.properties

Required RBAC: Source → data_dcp_reader on bucket; Sink → data_writer on collection.

---

Source Connector — Couchbase → Kafka

name=couchbase-source
connector.class=com.couchbase.connect.kafka.CouchbaseSourceConnector
tasks.max=2

couchbase.seed.nodes=localhost
couchbase.bucket=myapp
couchbase.username=kafka-source-user
couchbase.password=secret

# Stream specific collections (Server 7.0+)
couchbase.collections=_default.orders,_default.products

# Topic routing (default: ${bucket}.${scope}.${collection})
couchbase.topic[_default.orders]=myapp.orders
couchbase.topic[_default.products]=myapp.products

# Message format
couchbase.source.handler=com.couchbase.connect.kafka.handler.source.RawJsonSourceHandler
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter

# Where to start: SAVED_OFFSET_OR_BEGINNING | SAVED_OFFSET_OR_NOW | BEGINNING | NOW
couchbase.stream.from=SAVED_OFFSET_OR_BEGINNING

Source handlers

Handler	Output
RawJsonSourceHandler	Raw JSON bytes — identical to Couchbase doc
RawJsonWithMetadataSourceHandler	{event, key, cas, content} envelope — includes event type
DefaultSchemaSourceHandler	Avro/JSON with schema (Schema Registry)

Filtering and metadata

# JSONPath filter — only stream matching documents
couchbase.jsonpath.filter=$.type == "order"
couchbase.jsonpath.filter[_default.events]=$.severity == "critical"

# Attach Couchbase metadata as Kafka record headers
couchbase.headers=bucket,collection,key,cas,seqno,expiry

# Drop deletion events
transforms=ignoreDeletes
transforms.ignoreDeletes.type=com.couchbase.connect.kafka.transform.DropIfNullValue

---

Sink Connector — Kafka → Couchbase

name=couchbase-sink
connector.class=com.couchbase.connect.kafka.CouchbaseSinkConnector
tasks.max=2
topics=myapp.orders,myapp.products

couchbase.seed.nodes=localhost
couchbase.bucket=myapp
couchbase.username=kafka-sink-user
couchbase.password=secret

# Route topics to collections
couchbase.default.collection[myapp.orders]=_default.orders
couchbase.default.collection[myapp.products]=_default.products

# Document ID from body field (default: Kafka record key)
couchbase.document.id=${/id}
couchbase.remove.document.id=true

# Retry failed writes for 30 minutes
couchbase.retry.timeout=30m

# TTL per topic
couchbase.document.expiration[myapp.sessions]=24h

Sink handlers

Handler	Use when
UpsertSinkHandler (default)	Replace entire document
SubDocumentSinkHandler	Update specific fields only
N1qlSinkHandler	Complex conditional logic, MERGE

A Kafka record with null value deletes the Couchbase document matching the record key.

---

DCP Internals

DCP streams changes from 1024 vBuckets. Each tasks.max task handles a subset. The connector checkpoints sequence numbers per vBucket to Kafka's offset store.

Rollback: if Couchbase detects the connector's saved sequence is ahead of vBucket history (e.g., after failover), it rolls back. The connector handles this automatically but may re-deliver events — make consumers idempotent.

Scaling: set tasks.max ≈ number of Couchbase nodes for best parallelism.

Direct DCP streaming (without Kafka)

Use the couchbase-dcp-client Java library to consume DCP directly — useful for custom change-data-capture pipelines, audit trails, or cache invalidation.

<!-- Maven dependency -->
<dependency>
  <groupId>com.couchbase.client</groupId>
  <artifactId>dcp-client</artifactId>
  <version>0.49.0</version>
</dependency>

import com.couchbase.client.dcp.*;
import com.couchbase.client.dcp.message.*;

Client client = Client.builder()
    .seedNodes("localhost")
    .bucket("myapp")
    .credentials("Administrator", "password")
    .build();

// Handle control events (snapshot markers, stream end)
client.controlEventHandler((flowController, event) -> {
    if (DcpSnapshotMarkerRequest.is(event)) {
        flowController.ack(event);
    }
    event.release();
});

// Handle data events (mutations, deletions, expirations)
client.dataEventHandler((flowController, event) -> {
    if (DcpMutationMessage.is(event)) {
        String key = DcpMutationMessage.keyString(event);
        System.out.println("Mutation: " + key);
    } else if (DcpDeletionMessage.is(event)) {
        String key = DcpDeletionMessage.keyString(event);
        System.out.println("Deletion: " + key);
    }
    flowController.ack(event);
    event.release();
});

client.connect().block();
// Stream from current state (tail) — use StreamFrom.BEGINNING to replay all history
client.initializeState(StreamFrom.NOW, StreamTo.INFINITY).block();
client.startStreaming().block();

Always call flowController.ack(event) and event.release() to prevent memory leaks and flow-control stalls.

---

Transactional Outbox Pattern

Write business document + outbox event atomically in a Couchbase transaction; stream the outbox collection to Kafka.

// Node.js — atomic write
await cluster.transactions().run(async (ctx) => {
    const orderId = `order::${uuid()}`;
    await ctx.insert(orders, orderId, { type: "order", userId, items, total });
    await ctx.insert(outbox, `outbox::${uuid()}`, {
        type: "OrderCreated", aggregateId: orderId,
        payload: { userId, total }, createdAt: new Date().toISOString()
    });
});

# Source connector — stream only the outbox collection
couchbase.collections=_default.outbox
couchbase.topic[_default.outbox]=domain-events
couchbase.source.handler=com.couchbase.connect.kafka.handler.source.RawJsonWithMetadataSourceHandler

If the app crashes before writing, neither document exists. If Kafka delivery fails, the outbox event remains and will be re-streamed on connector restart.

---

Running the Connector

# Standalone (development)
env CLASSPATH="$KAFKA_CONNECT_COUCHBASE_HOME/lib/*" \
  connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties \
                        quickstart-couchbase-source.properties

# Distributed (production) — REST API
curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d @config.json
curl http://localhost:8083/connectors/couchbase-source/status
curl -X POST http://localhost:8083/connectors/couchbase-source/tasks/0/restart

---

Troubleshooting

Problem	Cause	Fix
Connector not starting	Wrong bucket or missing data_dcp_reader	Check connect.log for AUTH_ERROR
No messages in topic	SAVED_OFFSET_OR_NOW with existing offsets	Use BEGINNING to force full re-stream
Sink writing wrong collection	Topic name mismatch in override	Ensure topic name in [topic] override matches topics list exactly
High lag	Too few tasks	Increase tasks.max up to node count
Rollback to zero on restart	Lost saved offsets	Set couchbase.black.hole.topic and couchbase.initial.offset.topic