Couchbase Source Connector - AutoMQ Documentation

Overview

The Couchbase Source Connector reads document mutation, deletion, and expiration events from the DCP change stream of a Couchbase bucket, converts them into Kafka source records, and routes the records to Kafka topics according to the configuration. One connector instance connects to one bucket. It can read all scopes and collections, or be limited to one scope or a set of scope.collection values. The default handler outputs records with Connect schemas. You can also select the Raw JSON handler to write JSON mutations to Kafka as raw bytes, or select the Raw JSON handler with metadata. Records can carry headers such as document ID, bucket, scope, collection, vBucket, sequence number, CAS, and expiration time, which helps downstream systems implement idempotent processing and audit indexes. The connector stores source offsets for each vBucket and resumes from the saved position after a restart or task reassignment. Design the normal single-record path for at-least-once delivery: if a failure occurs after Kafka writes a record but before the offset is persisted, earlier records may appear again. Ordering is guaranteed only within the same source vBucket; there is no global ordering guarantee across vBuckets, Tasks, or Topics.

Prerequisites

The Couchbase account must be able to access the target bucket and have the permissions required to read the DCP change stream; when selecting scopes or collections, Couchbase Server must support collection streams.
If TLS is enabled, the Worker must be able to read the configured PEM CA file or Java keystore. If client certificate authentication is used, prepare readable client certificate files and their password.
If the default persistence check using couchbase.persistence.polling.interval is enabled, the source bucket must support persistence; an ephemeral bucket must set this configuration to 0.
The target Topic, black-hole Topic, initial-offset Topic, or schema-failure Topic must be prepared in advance and allow the Connect Worker to write. The connector does not create these Topics for you.
When RawJsonSourceHandler or RawJsonWithMetadataSourceHandler is selected, the Worker or Connector must use a ByteArrayConverter that matches the raw-byte output.

License

Licensed under Apache License 2.0.

Quick Start

Prepare a Connect Cluster, Kafka cluster, and accessible Couchbase bucket, and confirm network connectivity and access permissions from the Worker to Couchbase, Kafka, and the target Topic. The following Connector properties configuration uses placeholders for addresses, resource names, and credentials. For general instructions to create and manage a Connector, see Manage Connectors.

connector.class=com.couchbase.connect.kafka.CouchbaseSourceConnector
couchbase.seed.nodes=<couchbase-kv-host>
couchbase.username=<couchbase-username>
couchbase.password=<couchbase-password>
couchbase.bucket=<bucket-name>
couchbase.topic=<kafka-topic>
couchbase.source.handler=com.couchbase.connect.kafka.handler.source.RawJsonSourceHandler
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
key.converter=org.apache.kafka.connect.storage.StringConverter

Set couchbase.seed.nodes to one or more Couchbase KV node addresses, couchbase.bucket to the source bucket, and couchbase.topic to a prepared Kafka Topic. This example publishes raw bytes for JSON mutations; the value is null for deletion and expiration events. Do not put real passwords in documentation, logs, or shared configuration in production; use secure configuration injection when applying the configuration.

Configuration

Connection and Authentication

`couchbase.seed.nodes`

The list of seed node addresses for the Couchbase cluster.

Type: list
Default: None
Importance: High
Valid values / notes: Required. Use comma-separated node addresses; custom ports should be KV ports, normally non-TLS 11210 or TLS 11207.
Required: Yes

`couchbase.username`

The username used to connect to Couchbase.

Type: string
Default: None
Importance: High
Valid values / notes: Required. When client certificate authentication is used, the runtime can ignore this value, but the configuration item must still be provided to satisfy the configuration definition.
Required: Yes

`couchbase.password`

The password used to connect to Couchbase.

Type: password
Default: None
Importance: High
Valid values / notes: Required. It can be provided through KAFKA_COUCHBASE_PASSWORD when the configuration is read; when client certificate authentication is used, the runtime can ignore this value, but the configuration item must still be provided.
Required: Yes

`couchbase.bucket`

The Couchbase bucket whose change stream is read.

Type: string
Default: Empty string
Importance: High
Valid values / notes: Required at runtime and cannot be empty. One Connector instance connects to one bucket only.
Required: Yes

`couchbase.network`

Selects the Couchbase SDK network resolution mode.

Type: string
Default: auto
Importance: Medium
Valid values / notes: Case-sensitive. Allowed values are auto, default, and external.

`couchbase.bootstrap.timeout`

The timeout used when starting the connector and establishing the DCP socket.

Type: string
Default: 30s
Importance: Medium
Valid values / notes: Use an integer followed by ms, s, m, h, or d, for example 10s.

TLS and Client Certificates

`couchbase.enable.tls`

Whether to enable the Couchbase TLS connection.

Type: boolean
Default: false
Importance: Medium
Valid values / notes: When enabled, the trust certificate, hostname verification, and optional client certificate settings are used; the default CA is used when custom trust material is not specified.

`couchbase.enable.hostname.verification`

Whether to verify the hostname in the TLS certificate.

Type: boolean
Default: true
Importance: Medium
Valid values / notes: Applies only to TLS connections. Do not disable hostname verification unless there is a clear certificate deployment reason.

`couchbase.trust.store.path`

The absolute path to the Java keystore trust store.

Type: string
Default: Empty string
Importance: Medium
Valid values / notes: Use together with couchbase.trust.store.password; you can also use couchbase.trust.certificate.path instead.

`couchbase.trust.store.password`

The password for the Java keystore trust store.

Type: password
Default: Empty string
Importance: Medium
Valid values / notes: Provide it when couchbase.trust.store.path is configured. It can be overridden through KAFKA_COUCHBASE_TRUST_STORE_PASSWORD.

`couchbase.trust.certificate.path`

The absolute path to a PEM-format CA certificate.

Type: string
Default: Empty string
Importance: Medium
Valid values / notes: Use either this setting or the trust store. The Worker process must be able to read the file.

`couchbase.client.certificate.path`

The path to the client certificate keystore or PKCS12 bundle.

Type: string
Default: Empty string
Importance: Medium
Valid values / notes: A non-empty value enables client certificate authentication and ignores the runtime username and password values.

`couchbase.client.certificate.password`

The password for the client certificate file.

Type: password
Default: Empty string
Importance: Medium
Valid values / notes: Use together with couchbase.client.certificate.path. It can be overridden through KAFKA_COUCHBASE_CLIENT_CERTIFICATE_PASSWORD.

Topic Routing and Record Processing

`couchbase.topic`

The destination Kafka Topic template used when no collection-level override is configured.

Type: string
Default: ${bucket}.${scope}.${collection}
Importance: Medium
Valid values / notes: Supports the ${bucket}, ${scope}, and ${collection} placeholders. You can also use couchbase.topic[scope.collection] to override the Topic for a collection.

`couchbase.collection.to.topic`

Configures collection-to-Topic mappings in the form scope.collection=topic.

Type: list
Default: Empty list
Importance: Medium
Valid values / notes: Deprecated. Use couchbase.topic[scope.collection] for new configurations; unmapped collections fall back to couchbase.topic.
Deprecated: Yes
Replacement: couchbase.topic[scope.collection]

`couchbase.source.handler`

The handler class that converts Couchbase document events into Kafka source records.

Type: class
Default: None
Importance: Medium
Valid values / notes: Required. The class must implement SourceHandler or MultiSourceHandler and be instantiable. When RawJsonSourceHandler is used, value.converter should be set to org.apache.kafka.connect.converters.ByteArrayConverter.
Required: Yes

`couchbase.headers`

Selects Couchbase metadata headers to attach to Kafka records.

Type: list
Default: Empty list
Importance: Medium
Valid values / notes: Allowed values are bucket, scope, collection, key, qualifiedKey, cas, partition, partitionUuid, seqno, rev, and expiry.

`couchbase.header.name.prefix`

Adds a prefix to the names of the headers selected by couchbase.headers.

Type: string
Default: couchbase.
Importance: Medium
Valid values / notes: The prefix is added before every selected header name.

`couchbase.event.filter`

The Filter class used to filter Couchbase events.

Type: class
Default: com.couchbase.connect.kafka.filter.AllPassFilter
Importance: Medium
Valid values / notes: The class must be instantiable and implement Filter. The default filter excludes system-scope events and transaction metadata documents whose keys start with _txn:.

`couchbase.jsonpath.filter`

Filters mutations according to document JSON content.

Type: string
Default: Empty string
Importance: Medium
Valid values / notes: An empty value disables JSONPath filtering; a non-empty value must be a parseable JSONPath. It runs before the event Filter and source handler. You can override it with couchbase.jsonpath.filter[scope.collection].

`couchbase.no.value`

Whether DCP should omit the document body of mutations.

Type: boolean
Default: false
Importance: Medium
Valid values / notes: When enabled, the key and metadata are retained but the body is not provided. JSONPath, Filters, Handlers, and schema mappings that depend on the body cannot generate body content.

`couchbase.xattrs`

Whether to request Couchbase extended attributes.

Type: boolean
Default: false
Importance: Medium
Valid values / notes: When enabled, a Filter or custom source handler can read xattrs. Enable it only when this metadata is actually needed.

Start Position and Offset

`couchbase.stream.from`

Sets the starting point for the first startup or for a vBucket without a saved offset.

Type: string
Default: SAVED_OFFSET_OR_BEGINNING
Importance: Medium
Valid values / notes: Case-sensitive. Allowed values are SAVED_OFFSET_OR_BEGINNING, SAVED_OFFSET_OR_NOW, BEGINNING, and NOW. BEGINNING is limited by Couchbase’s retained history boundary.

`couchbase.initial.offset.topic`

The Topic to which synthetic records for the initial position of a vBucket without a saved offset are published.

Type: string
Default: Empty string
Importance: Medium
Valid values / notes: Relevant only when couchbase.stream.from=SAVED_OFFSET_OR_NOW; when couchbase.black.hole.topic is configured, use the same Topic. This Topic is not a business-data Topic.

`couchbase.black.hole.topic`

Receives synthetic records for events ignored by the Filter or Handler so that their source offsets can be committed.

Type: string
Default: Empty string
Importance: Medium
Valid values / notes: When non-empty, a small placeholder record is sent for each ignored event. Configure a short retention period and small segments for this Topic according to business needs. It is not a business DLQ.

`couchbase.connector.name.in.offsets`

Whether to include the Connector name in the source offset identity.

Type: boolean
Default: false
Importance: Medium
Valid values / notes: Deprecated. Keep it false for new deployments; set it to true only for compatibility with older deployments that stored offsets with the Connector name.
Deprecated: Yes
Replacement: Keep it false for new deployments and use Kafka Connect’s default Connector offset isolation.

Batching and DCP

`couchbase.batch.size.max`

The maximum number of SourceRecords returned by a single poll.

Type: int
Default: 2000
Importance: Medium
Valid values / notes: Limits only the size of a single poll batch. It does not limit the Task’s internal event queue or in-flight Kafka producer data. Adjust it together with Worker throughput and heap memory.

`couchbase.compression`

Selects the DCP transport compression mode.

Type: string
Default: ENABLED
Importance: Medium
Valid values / notes: Case-sensitive. Allowed values are DISABLED, FORCED, and ENABLED. It affects only transmission from Couchbase to the Connector and does not change Kafka record compression.

`couchbase.persistence.polling.interval`

The polling interval for waiting until Couchbase changes meet the persistence condition before publishing them.

Type: string
Default: 100ms
Importance: Medium
Valid values / notes: Use an integer followed by ms, s, m, h, or d; set it to 0 to disable the check. An ephemeral bucket must set it to 0, otherwise events may not be published.

`couchbase.flow.control.buffer`

The DCP flow-control buffer size used by each Task on each Couchbase node.

Type: string
Default: 16m
Importance: Medium
Valid values / notes: Supports b, k, m, and g. The buffer scales with the number of Tasks and nodes, so increasing it increases memory usage. A common tuning range is 10m to 50m, but actual load must still be evaluated.

Schema Processing

`couchbase.value.schema`

The Avro record schema JSON used by source handlers that support schemas.

Type: string
Default: Empty string
Importance: Medium
Valid values / notes: Applies only to handlers that support this setting. ConfigurableSchemaSourceHandler requires a parseable Avro record schema. You can override it per collection with couchbase.value.schema[scope.collection].

`couchbase.schema.failure.action`

The action taken by the Schema Registry handler when a schema is missing or does not match.

Type: string
Default: TERMINATE
Importance: Medium
Valid values / notes: Case-sensitive. Allowed values are TERMINATE, DROP, and DLQ. Applies only to Schema Registry handlers that support this option.

`couchbase.dlq.topic`

The destination Topic used by the Schema Registry handler when couchbase.schema.failure.action=DLQ.

Type: string
Default: couchbase.dlq
Importance: Medium
Valid values / notes: Used only for forwarding schema failures from that handler; it is not the same as the Kafka Connect generic error-handling DLQ. The Worker must be allowed to write to the target Topic.

Logging and Diagnostics

`couchbase.log.redaction`

Controls the level of sensitive-information redaction in Couchbase client logs.

Type: string
Default: NONE
Importance: Medium
Valid values / notes: Case-sensitive. Allowed values are NONE, PARTIAL, and FULL. Select one according to your organization’s log security requirements.

`couchbase.log.document.lifecycle`

Whether to promote document lifecycle milestones from DEBUG to INFO.

Type: boolean
Default: false
Importance: Medium
Valid values / notes: When enabled, it may generate a high volume of logs and is suitable for short-term investigation of an individual document’s flow.

`couchbase.metrics.interval`

The interval for periodically writing Connector metrics to the log.

Type: string
Default: 10m
Importance: Medium
Valid values / notes: Use an integer followed by ms, s, m, h, or d; set it to 0 to disable metric logging. This is a diagnostic setting without a stability commitment.

`couchbase.enable.dcp.trace`

Whether to enable detailed DCP trace logging.

Type: boolean
Default: false
Importance: Medium
Valid values / notes: When enabled, detailed DCP diagnostics are logged at INFO level and couchbase.dcp.trace.document.id.regex is enabled. Use it only for a short period during troubleshooting.

`couchbase.dcp.trace.document.id.regex`

The regular expression for document IDs to record in the DCP trace.

Type: string
Default: .*
Importance: Medium
Valid values / notes: Used only when couchbase.enable.dcp.trace=true; the Java regular expression is compiled when the connector starts.

Collection Scope

`couchbase.scope`

Selects all collections in one scope.

Type: string
Default: Empty string
Importance: Medium
Valid values / notes: Requires Couchbase Server 7.0 or later. Mutually exclusive with couchbase.collections; when both are empty, all scopes and collections are read.

`couchbase.collections`

Selects a list of specific collections to read.

Type: list
Default: Empty list
Importance: Medium
Valid values / notes: Use comma-separated scope.collection names. Requires Couchbase Server 7.0 or later. Mutually exclusive with couchbase.scope; when both are empty, all scopes and collections are read.

Dynamic Collection Context Configuration

`couchbase.topic[scope.collection]`

Overrides the couchbase.topic template for a specified scope.collection.

Type: string
Default: Inherits couchbase.topic ${bucket}.${scope}.${collection}
Importance: Medium
Valid values / notes: Registered only when a correctly formatted bracketed property appears in the submitted configuration. The bracket contents must be a qualified scope and collection; unmatched collections fall back to the base configuration.

`couchbase.jsonpath.filter[scope.collection]`

Overrides the JSONPath filter expression for a specified scope.collection.

Type: string
Default: Inherits the empty string from couchbase.jsonpath.filter
Importance: Medium
Valid values / notes: Every override must be a parseable JSONPath; unmatched collections use the base configuration.

`couchbase.value.schema[scope.collection]`

Overrides the schema JSON used by the handler for a specified scope.collection.

Type: string
Default: Inherits the empty string from couchbase.value.schema
Importance: Medium
Valid values / notes: Applies only to source handlers that support this setting; every override must meet the schema requirements of that handler.

Kafka Connect Framework

`connector.class`

Selects the Kafka Connect Source plugin class to run.

Type: string
Default: None
Importance: High
Valid values / notes: Required. Use com.couchbase.connect.kafka.CouchbaseSourceConnector.
Required: Yes

`tasks.max`

The maximum number of Tasks that this Connector may create.

Type: int
Default: 1
Importance: High
Valid values / notes: Must be at least 1. The connector splits work by Couchbase vBucket, and the actual number of non-empty Tasks cannot exceed the number of vBuckets. Multiple Tasks from the same Connector do not consume the same vBucket in parallel.

`key.converter`

The Converter used to serialize the Source record key.

Type: class
Default: null (explicit value; inherits the Worker configuration)
Importance: Low
Valid values / notes: Must be an instantiable Kafka Connect Converter. A Source handler usually uses the document ID as the key. The example uses org.apache.kafka.connect.storage.StringConverter.

`value.converter`

The Converter used to serialize the Source record value.

Type: class
Default: null (explicit value; inherits the Worker configuration)
Importance: Low
Valid values / notes: Must be an instantiable Kafka Connect Converter. The Raw JSON handler requires org.apache.kafka.connect.converters.ByteArrayConverter; for other handlers, select a Converter that matches the output schema.

Best Practices

Receive Only Changes Enabled After Creating a Real-Time Pipeline

Use this pattern when creating a new real-time event pipeline that does not need to replay historical changes still retained by Couchbase. The goal is for vBuckets with saved offsets to recover normally, for vBuckets without saved offsets to start from the position when the Connector is enabled, and for that initial position to enter the Kafka Connect offset commit path.

connector.class=com.couchbase.connect.kafka.CouchbaseSourceConnector
couchbase.seed.nodes=<couchbase-kv-host>
couchbase.username=<couchbase-username>
couchbase.password=<couchbase-password>
couchbase.bucket=<bucket-name>
couchbase.topic=<kafka-topic>
couchbase.source.handler=com.couchbase.connect.kafka.handler.source.RawJsonSourceHandler
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
couchbase.stream.from=SAVED_OFFSET_OR_NOW
couchbase.initial.offset.topic=<offset-topic>

Key points: Create <offset-topic> in advance and allow the Worker to write to it. SAVED_OFFSET_OR_NOW does not replay existing history for vBuckets without saved offsets; couchbase.initial.offset.topic writes a synthetic record for the initial position, which becomes a recovery point after that record and the source offset are committed successfully. It is not a business data Topic and should not be processed by business consumers.

Limit the Source by Collection and Route to Different Topics

Use this pattern when only some collections in a bucket should enter Kafka and different business collections need to be isolated into different Topics. Limit the source range when the DCP stream is established, and use collection context overrides for Topics so that unrelated collections are not first read and then filtered on the Kafka side.

connector.class=com.couchbase.connect.kafka.CouchbaseSourceConnector
couchbase.seed.nodes=<couchbase-kv-host>
couchbase.username=<couchbase-username>
couchbase.password=<couchbase-password>
couchbase.bucket=<bucket-name>
couchbase.collections=orders.pending,orders.completed
couchbase.topic=${scope}-${collection}
couchbase.source.handler=com.couchbase.connect.kafka.handler.source.RawJsonWithMetadataSourceHandler
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter

Key points: couchbase.collections uses complete scope.collection names. Do not use wildcards or also configure the mutually exclusive couchbase.scope. The Topic template routes the two collections to orders-pending and orders-completed, respectively; prepare the corresponding Topics in advance. If only individual collections need exception names, use the couchbase.topic[scope.collection] override described in the Configuration section. Raw JSON output with metadata allows consumers to use the raw body and event context together.

Choose the Trade-Off Between Failure Consistency and Throughput

Use this pattern when a persistent bucket needs to reduce the likelihood that source-side failover rollback events enter Kafka, or when an ephemeral bucket needs events to be published immediately. Choose the persistence check according to the bucket type and acceptable latency, and use the DCP flow-control buffer to control throughput and memory usage.

connector.class=com.couchbase.connect.kafka.CouchbaseSourceConnector
couchbase.seed.nodes=<couchbase-kv-host>
couchbase.username=<couchbase-username>
couchbase.password=<couchbase-password>
couchbase.bucket=<bucket-name>
couchbase.topic=<kafka-topic>
couchbase.source.handler=com.couchbase.connect.kafka.handler.source.RawJsonSourceHandler
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
couchbase.persistence.polling.interval=100ms
couchbase.flow.control.buffer=16m

Key points: The persistence check adds latency, network traffic, and memory overhead, but can reduce the risk of alternate-history events after source-side failover. Setting it to 0 prioritizes throughput and low latency, but downstream systems must accept that risk. An ephemeral bucket must use 0, otherwise the persistence condition cannot be met. The buffer scales with the number of Tasks and Couchbase nodes, so heap memory cannot be estimated from the single configuration value alone.

Monitoring

What to Monitor

Monitor the health and restart counts of the Kafka Connect Worker, Connector, and Tasks. Observe Source throughput, end-to-end latency, poll latency, source offset commit progress, errors, and retries. Also monitor Worker JVM heap usage, GC, threads, and CPU. If error handling or the Schema Registry handler’s dedicated failure Topic is enabled, monitor writes and backlog for the corresponding DLQ or failure Topic, and analyze them together with target Topic production rate, consumer lag, and Couchbase source load.

Import the Grafana Dashboard

Download the AutoMQ Connect Cluster Dashboard, select in Grafana the Prometheus data source connected to Connect Worker metrics, confirm that metric labels match the dashboard variables, and import the JSON through Grafana. This dashboard covers Kafka Connect cluster-level metrics; use Worker logs and Couchbase monitoring for Connector-specific diagnostics.

Limitations

One Connector instance can connect to only one Couchbase bucket; create separate Connector instances for multiple buckets.
The connector guarantees ordering only within the same source vBucket. It does not guarantee global ordering across vBuckets, Tasks, Topics, or collections.
Design the normal single-record path for at-least-once delivery; failures after Kafka writes but before source offset persistence, as well as Couchbase failover rewinds, may produce duplicate records.
The Task’s internal event queue has no hard limit based on element count; couchbase.flow.control.buffer controls DCP byte flow and is not a JVM queue capacity limit.
The connector does not provide unified count or backoff-retry settings for connection, Filter, Handler, schema, and producer failures; recovery behavior depends on the DCP client and Kafka Connect Task lifecycle.

FAQ

The Connector has started, but there are no business messages in Kafka. What should I check?

First check the Couchbase address, bucket, account permissions, and write permission for the target Topic. Then confirm that couchbase.source.handler can be instantiated and that couchbase.stream.from matches the expected starting point. If couchbase.scope, couchbase.collections, or JSONPath is configured, confirm that the target documents are in the selected range and satisfy the expression. The default Filter also excludes system scopes and _txn: transaction metadata documents. Finally, check Task logs for DCP connection errors and Kafka producer errors.

Why do records that were already published appear again after a restart?

The source offset is persisted by Kafka Connect only after the Kafka record has been written successfully. If the Worker stops after the write but before the offset commit, recovery may replay from an earlier position. A Couchbase failover rewind may also cause duplicates. Consumers should implement idempotent processing using stable document IDs, vBucket and seqno metadata, and should avoid changing the Connector identity or compatibility settings for saved offsets without a clear reason.

Why is the body empty after setting `couchbase.no.value=true`?

This setting causes DCP to omit the mutation body and retain only the key and metadata. It is suitable when processing requires only event identity or metadata. When business-field filtering, JSON generation, or schema-record construction is required, restore it to false and confirm that couchbase.jsonpath.filter, the Filter, and the source handler do not depend on the omitted body.

Why are events delayed, or are there no events at all, after persistence polling is enabled?

Persistence polling waits for source changes to meet the persistence condition, so it adds latency and memory usage. For a persistent bucket, adjust couchbase.persistence.polling.interval according to consistency requirements. An ephemeral bucket has no persistence condition and must set this value to 0. Also check the DCP flow-control buffer, Task count, and Couchbase node status.

Why are records from different collections entering the same Topic?

Check whether the couchbase.topic template includes ${scope} and ${collection}, and whether collection-level properties use the complete couchbase.topic[scope.collection] format. A collection-level override applies only to the qualified name inside the brackets; unmatched collections fall back to the base Topic. The connector does not automatically create or validate Topic names.

What happens when a schema failure occurs with a Schema Registry handler?

Check couchbase.schema.failure.action. TERMINATE fails the Task, DROP discards the mismatched event, and DLQ sends a diagnostic record to couchbase.dlq.topic. These settings apply only to Schema Registry handlers that support them. The target Topic, value Converter, and Schema Registry access must also match; this is not the same as the Kafka Connect generic DLQ.

​Overview

​Prerequisites

​License

​Quick Start

​Configuration

​Connection and Authentication

​couchbase.seed.nodes

​couchbase.username

​couchbase.password

​couchbase.bucket

​couchbase.network

​couchbase.bootstrap.timeout

​TLS and Client Certificates

​couchbase.enable.tls

​couchbase.enable.hostname.verification

​couchbase.trust.store.path

​couchbase.trust.store.password

​couchbase.trust.certificate.path

​couchbase.client.certificate.path

​couchbase.client.certificate.password

​Topic Routing and Record Processing

​couchbase.topic

​couchbase.collection.to.topic

​couchbase.source.handler

​couchbase.headers

​couchbase.header.name.prefix

​couchbase.event.filter

​couchbase.jsonpath.filter

​couchbase.no.value

​couchbase.xattrs

​Start Position and Offset

​couchbase.stream.from

​couchbase.initial.offset.topic

​couchbase.black.hole.topic

​couchbase.connector.name.in.offsets

​Batching and DCP

​couchbase.batch.size.max

​couchbase.compression

​couchbase.persistence.polling.interval

​couchbase.flow.control.buffer

​Schema Processing

​couchbase.value.schema

​couchbase.schema.failure.action

​couchbase.dlq.topic

​Logging and Diagnostics

​couchbase.log.redaction

​couchbase.log.document.lifecycle

​couchbase.metrics.interval

​couchbase.enable.dcp.trace

​couchbase.dcp.trace.document.id.regex

​Collection Scope

​couchbase.scope

​couchbase.collections

​Dynamic Collection Context Configuration

​couchbase.topic[scope.collection]

​couchbase.jsonpath.filter[scope.collection]

​couchbase.value.schema[scope.collection]

​Kafka Connect Framework

​connector.class

​tasks.max

​key.converter

​value.converter

​Best Practices

​Receive Only Changes Enabled After Creating a Real-Time Pipeline

​Limit the Source by Collection and Route to Different Topics

​Choose the Trade-Off Between Failure Consistency and Throughput

​Monitoring

​What to Monitor

​Import the Grafana Dashboard

​Limitations

​FAQ

​The Connector has started, but there are no business messages in Kafka. What should I check?

​Why do records that were already published appear again after a restart?

​Why is the body empty after setting couchbase.no.value=true?

​Why are events delayed, or are there no events at all, after persistence polling is enabled?

​Why are records from different collections entering the same Topic?

​What happens when a schema failure occurs with a Schema Registry handler?