Couchbase Kafka Connector issue

pramod.k.kaza · January 30, 2020, 6:05pm

Hi,
I am using couchbase kafka connector to capture CDC events to Confluent kafka topic …I get configuration error below
Any help is greatly appreciated
[2020-01-30 17:52:30,729] ERROR WorkerConnector{id=test-couchbase-source} Error while starting connector (org.apache.kafka.connect.runtime.WorkerConnector)
kafka-connect_1 | org.apache.kafka.connect.errors.ConnectException: Cannot fetch configuration for bucket CentralDB
kafka-connect_1 | at com.couchbase.connect.kafka.CouchbaseSourceConnector.start(CouchbaseSourceConnector.java:60)
kafka-connect_1 | at org.apache.kafka.connect.runtime.WorkerConnector.doStart(WorkerConnector.java:111)
kafka-connect_1 | at org.apache.kafka.connect.runtime.WorkerConnector.start(WorkerConnector.java:136)
kafka-connect_1 | at org.apache.kafka.connect.runtime.WorkerConnector.transitionTo(WorkerConnector.java:196)
kafka-connect_1 | at org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:231)
kafka-connect_1 | at org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:908)
kafka-connect_1 | at org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1300(DistributedHerder.java:110)
kafka-connect_1 | at org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:924)
kafka-connect_1 | at org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:920)
kafka-connect_1 | at java.util.concurrent.FutureTask.run(FutureTask.java:266)
kafka-connect_1 | at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
kafka-connect_1 | at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
kafka-connect_1 | at java.lang.Thread.run(Thread.java:748)

david.nault · January 30, 2020, 9:22pm

Hi Pramod,

The “Cannot fetch configuration for bucket” error occurs when the connector is unable to connect to any of the seed nodes on port 8091 using the credentials from the config file. I think it might also happen if the bucket does not exist. There might be more info about the root cause earlier in the log.

I’d recommend double-checking the following config properties to make sure they’re correct for your environment. Try using admin credentials to see if that’s the issue (then scale back as appropriate).

connection.cluster_address
connection.username
connection.password
connection.bucket

Thanks,
David

pramod.k.kaza · January 31, 2020, 5:54pm

the connection.cluster_address": “127.0.0.1” . I am using it in local . Just to give more background , I am using a dockerized confluent kafka which is pulling the couchbase kafka jar as plugin. confluent provides a REST API to inject configuration to connectors. I am using POST connectors API (http://localhost:18083/connectors) and gave below config to inject
{
“name”: “test-couchbase-source”,
“config”: {
“name”: “test-couchbase-source”,
“connector.class”: “com.couchbase.connect.kafka.CouchbaseSourceConnector”,
“tasks.max”: “2”,
“topic.name”: “test-source”,
“connection.cluster_address”: “127.0.0.1”,
“connection.timeout.ms”: “2000”,
“connection.bucket”: “CentralDB”,
“connection.username”: “Administrator”,
“connection.password”: “password”,
“use_snapshots”:“false”,
“dcp.message.converter.class”: “com.couchbase.connect.kafka.handler.source.DefaultSchemaSourceHandler”,
“event.filter.class”: “com.couchbase.connect.kafka.filter.AllPassFilter”,
“couchbase.stream_from”: “SAVED_OFFSET_OR_BEGINNING”,
“couchbase.compression”: “ENABLED”,
“couchbase.flow_control_buffer”: “128m”,
“couchbase.persistence_polling_interval”: “100ms”
}
}

However the configuration is not getting injected …Am i missing something ?

david.nault · January 31, 2020, 7:24pm

Can you please give some more details about how you’re making the POST request, what the response is, and any other details you think might be relevant?

There’s also a third-party Kafka Connect CLI tool that you might find useful. It takes a lot of the pain out of managing connectors via the HTTP interface.

pramod.k.kaza · January 31, 2020, 8:59pm

POST call from confluent kafka REST API - http://localhost:18083/connectors/
Payload to the REST API below
{
“name”: “test-couchbase-source”,
“config”: {
“name”: “test-couchbase-source”,
“connector.class”: “com.couchbase.connect.kafka.CouchbaseSourceConnector”,
“tasks.max”: “2”,
“topic.name”: “test-source”,
“connection.cluster_address”: “127.0.0.1”,
“connection.timeout.ms”: “2000”,
“connection.bucket”: “CentralDB”,
“connection.username”: “Administrator”,
“connection.password”: “password”,
“use_snapshots”:“false”,
“dcp.message.converter.class”: “com.couchbase.connect.kafka.handler.source.DefaultSchemaSourceHandler”,
“event.filter.class”: “com.couchbase.connect.kafka.filter.AllPassFilter”,
“couchbase.stream_from”: “SAVED_OFFSET_OR_BEGINNING”,
“couchbase.compression”: “ENABLED”,
“couchbase.flow_control_buffer”: “128m”,
“couchbase.persistence_polling_interval”: “100ms”
}
}
Docker compose file i run attached below

version: ‘2.1’

services:
zookeeper:
image: confluentinc/cp-zookeeper:5.2.1
hostname: zookeeper
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ports:
- 2181:2181

kafka:
image: confluentinc/cp-kafka:5.2.1
hostname: kafka
environment:
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka:19091,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9091
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL
KAFKA_BROKER_ID: 1
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
ports:
- 9091:9091
depends_on:
- zookeeper

kafka-connect:
# image: confluentinc/cp-kafka-connect:5.1.1
build: .
hostname: kafka-connect
environment:
CONNECT_BOOTSTRAP_SERVERS: “kafka:19091”
CONNECT_REST_PORT: 8083
CONNECT_GROUP_ID: kafka-connect-local
CONNECT_CONFIG_STORAGE_TOPIC: kafka-connect-local-config
CONNECT_OFFSET_STORAGE_TOPIC: kafka-connect-local-offset
CONNECT_STATUS_STORAGE_TOPIC: kafka-connect-local-status
CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
# CONNECT_OFFSET_STORAGE_PARTITIONS: 1
CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE: “true”
CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE: “true”
CONNECT_INTERNAL_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
CONNECT_INTERNAL_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
CONNECT_REST_ADVERTISED_HOST_NAME: ‘kafka-connect’
CONNECT_PLUGIN_PATH: /usr/share/java,/etc/kafka-connect/jars
ports:
- 18083:8083
depends_on:
- zookeeper
- kafka
**Couchbase is installed on my local , not a docker version **
The issue occurs when you try to inject the couchbase config above using REST API end point provided by confluent . I also suspect if the issue is with respect to what docker confluent considers as local host versus couchbase local host …any help greatly appreciated. is there a way i can call you ?

david.nault · January 31, 2020, 10:26pm

Ah, yes. That’s likely the problem. The connector running inside the Docker network environment cannot access Couchbase Server running outside the Docker environment. This limitation is not specific to the connector; it’s just how Docker networking operates. On certain operating systems there might be a way to configure Docker so the services can all see each other, but that’s not something I know about.

is there a way i can call you ?

For support beyond the “best effort” provided in the Forums, Couchbase Technical Support is probably the best escalation path.

As for solving the problem, the alternatives I’ve used in the past are:

A) Running Couchbase in Docker as well. It doesn’t need to be in the same docker compose file – just needs to use the same Docker network as the Kafka containers. Here’s a demo project that shows how do that.

B) Running Kafka on your local system without Docker. The Confluent CLI makes it really easy to start up the various Kafka services on your dev system.

Thanks,
David

pramod.k.kaza · February 6, 2020, 9:43pm

David
I some how had the connector working and its streaming CDC captures to my kafka topic on confluent kafka . However when i read the topic in consumer , the payload content is coming in byte format , not human readable . is there a setting i am missing to view it as JSON format ?

{ topic: ‘SharedKafka2_150119797_couchbase-connect-poc’,
value:
‘{“event”:“mutation”,“partition”:23,“key”:“asset::115217032”,“cas”:1581023894824091648,“bySeqno”:1407,“revSeqno”:5,“expiration”:0,“flags”:33554438,“lockTime”:0,“content”:“ewogICJhY2Nlc3NUb2tlbnMiOiB7CiAgICAiZTEiOiBbCiAgICAgICIwMDRiNzI3Ni1iNWQ5LTQyMmItOWJlMy0wN2IxODBlYzY5OTkiCiAgICBdLAogICAgImUyIjogWwogICAgICAiZjM4YWRlZDQtM2Y4Ni00N2EzLWFjNmYtZDdjYzA1ODRjZDg3IgogICAgXSwKICAgICJlMyI6IFsKICAgICAgImYwMzNjNDc4LTJjODUtNGNmNC05OGY2LTg2ODUyZDI3OGIyNyIKICAgIF0KICB9LAogICJhY3R1YWxFeGl0RGF0ZSI6ICIyMDA4LTA0LTE0VDA4OjAwOjAwLjAwMFoiLAogICJhbGlhc05hbWVzIjogW10sCiAgImFwcGxpY2F0aW9uRW50cnlNb2RlIjogIlByb2R1Y3Rpb24iLAogICJhcHBsaWNhdGlvblN0YXR1cyI6ICJBcHByb3ZlZCIsCiAgImFwcGxpY2F0aW9uVVJMIjogW10sCiAgImFzc2V0SWQiOiAxMTUyMTcwMzIsCiAgImJhbmtzIjogWwogICAgIkFFTkJOIgogIF0sCiAgImJ1c2luZXNzVEdvdmVybmFuY2VJbmRpY2F0b3IiOiBmYWxzZSwKICAiY2FwYWJpbGl0aWVzIjogW10sCiAgImNoYW5uZWxzIjogW10sCiAgImNyZWF0ZWREYXRlIjogIjIwMDgtMDQtMjZUMDg6MDA6MDAuMDAwWiIsCiAgImN1c3RvbWVycyI6IFtdLAogICJkYXRhRW50aXRpZXMiOiBbXSwKICAiZGF0YUludGVyZmFjZXMiOiBbXSwKICAiZGF0YUxldmVsSW1wYWN0IjogIk1lZGl1bSIsCiAgImRlc2NyaXB0aW9uIjogIlRyaWdnZXIgcmVwb3J0aW5nIHNlbmRzIGUtbWFpbHMgdG8gRVMgY2xpZW50IG1hbmFnZXJzIHRvIGFsZXJ0IGNoYW5nZXMgaW4gbWVyY2hhbnQgYXV0aCBzdWJtaXNzaW9ucyBjaGFyZ2Ugdm9sdW1lIGZyYXVkIGNoYXJnZWJhY2tzIHRoYXQgYXJlIG91dCBvZiBwYXR0ZXJuICBUaGUgcGxhdGZvcm0gcnVucyBvbiBBU1AgTkVUIGFuZCBNUyBTUUwgU2VydmVyIFJlcG9ydGluZyBTZXJ2aWNlcyBhbmQgdXNlcyBNSURXIGFzIHRoZSBwcmltYXJ5IGRhdGEgc291cmNlICBBdXRob3JpemF0aW9uIGRhdGEgaXMgbWFudWFsbHkgbG9hZGVkIGZyb20gQ0FTIHNkaHNoZHNqZGpzZGggOTk5OTk5ICIsCiAgImRvY1R5cGUiOiAiYXNzZXQiLAogICJleGl0UmVhc29uIjogIkRlY29tbWlzc2lvbiIsCiAgImZpbmFuY2lhbFJlc3BvbnNpYmlsaXR5IjogIiIsCiAgImhpc3RvcnkiOiBbCiAgICB7CiAgICAgICJjaGFuZ2VzIjogWwogICAgICAgIHsKICAgICAgICAgICJmaWVsZCI6ICJDcmVhdGVkIERhdGUiLAogICAgICAgICAgIm5ld1ZhbHVlIjogWwogICAgICAgICAgICAiMjAwOC0wNC0yNlQwODowMDowMC4wMDBaIgogICAgICAgICAgXSwKICAgICAgICAgICJvbGRWYWx1ZSI6IFtdCiAgICAgICAgfQogICAgICBdLAogICAgICAiZXZlbnRUeXBlIjogIk5ldyBBcHBsaWNhdGlvbiIsCiAgICAgICJsYXN0VXBkYXRlVGltZVN0YW1wIjogIjIwMDgtMDQtMjZUMDg6MDA6MDAuMDAwWiIsCiAgICAgICJsYXN0VXBkYXRlZEJ5IjogIkNlbnRyYWwiCiAgICB9CiAgXSwKICAiaG9zdGluZ1NpdGUiOiBbXSwKICAiaXNCcmFuZGVkIjogZmFsc2UsCiAgImtleSI6ICJhc3NldDo6MTE1MjE3MDMyIiwKICAibGFzdFVwZGF0ZVRpbWVTdGFtcCI6ICIyMDE2LTExLTE3VDAyOjMyOjUzLjY5MVoiLAogICJsZWdhY3lhcHBsaWNhdGlvbkVudHJ5TW9kZSI6ICJQcm9kdWN0aW9uIiwKICAibGVnYWN5bGlmZUN5Y2xlU3RhdHVzIjogIkV4aXQiLAogICJsaWZlQ3ljbGVTdGF0dXMiOiAiRXhpdCIsCiAgImxpbmVPZkJ1c2luZXNzIjoge30sCiAgIm1hbmFnZWRCeSI6IFsKICAgICJBbWVyaWNhbiBFeHByZXNzIFRlY2hub2xvZ3kiCiAgXSwKICAibmFtZSI6ICJFUyAtIFRSSUdHRVIgUkVQT1JUSU5HIC0gTkEiLAogICJub3RpZmljYXRpb25zIjogewogICAgImVtYWlsIjogdHJ1ZSwKICAgICJ3b3JrR3JvdXBOb3RpZmljYXRpb25MaXN0IjogW10KICB9LAogICJvd25lcnNoaXBJbmZvIjogewogICAgImFwcGxpY2F0aW9uTWFuYWdlciI6IHsKICAgICAgImJhbmQiOiA0MCwKICAgICAgImVtYWlsIjogIlNhbS5CYWtlckBhZXhwLmNvbSIsCiAgICAgICJmdWxsTmFtZSI6ICJTYW0gQmFrZXIiLAogICAgICAiZ3VpZCI6ICIwZGJjNmQ1ZWEzNWRkZTc1ODcxYTQzMzI0NTZjNTY4ZCIKICAgIH0sCiAgICAiYXBwbGljYXRpb25Pd25lciI6IHsKICAgICAgImJhbmQiOiA0MCwKICAgICAgImVtYWlsIjogIlNhbS5CYWtlckBhZXhwLmNvbSIsCiAgICAgICJmdWxsTmFtZSI6ICJTYW0gQmFrZXIiLAogICAgICAiZ3VpZCI6ICIwZGJjNmQ1ZWEzNWRkZTc1ODcxYTQzMzI0NTZjNTY4ZCIKICAgIH0sCiAgICAiYXBwbGljYXRpb25Pd25lckxlYWRlcjEiOiB7CiAgICAgICJlbWFpbCI6ICJzYXN0cnl2c20uZHVydmFzdWxhQGFleHAuY29tIiwKICAgICAgImd1aWQiOiAiODMzMWQ0MDQ0ZjIzNjY2YWIzYjg3MGZhY2ZlYTIxMGIiCiAgICB9LAogICAgImFwcGxpY2F0aW9uT3duZXJMZWFkZXIyIjogewogICAgICAiZW1haWwiOiAiTWlrZS5SdXR0bGVkZ2VAYWV4cC5jb20iLAogICAgICAiZ3VpZCI6ICIzYzMyMDIxZTdlZWRhZTVjNGJmYTYzMjM1OTljMTJhMSIKICAgIH0sCiAgICAiYnVzaW5lc3NPd25lciI6IHsKICAgICAgImVtYWlsIjogImxsb3lkLmcuY2F0b0BhZXhwLmNvbSIsCiAgICAgICJndWlkIjogIjE1NGRhZWEwZGFiN2JkMmQzNDQzOTg0Y2I5NGNhOWE5IgogICAgfSwKICAgICJidXNpbmVzc093bmVyTGVhZGVyMSI6IHsKICAgICAgImVtYWlsIjogImxsb3lkLmcuY2F0b0BhZXhwLmNvbSIsCiAgICAgICJndWlkIjogIjE1NGRhZWEwZGFiN2JkMmQzNDQzOTg0Y2I5NGNhOWE5IgogICAgfSwKICAgICJvd25lclNWUCI6IHsKICAgICAgImVtYWlsIjogIlBhdWwuRG90dGxlQGFleHAuY29tIiwKICAgICAgImd1aWQiOiAiMTEzNWFlNzJlZmViMjZkZGQ3MjE3ODRjMjhiZjYwMDYiCiAgICB9LAogICAgInByb2R1Y3Rpb25TdXBwb3J0T3duZXIiOiB7CiAgICAgICJiYW5kIjogNDAsCiAgICAgICJlbWFpbCI6ICJrZWl0aC5hLmJ1cm5zQGFleHAuY29tIiwKICAgICAgImZ1bGxOYW1lIjogIktlaXRoIEEgQnVybnMiLAogICAgICAiZ3VpZCI6ICIyNjM2NDIxNjhkMzg5NWZlMWY1OGE0YjkwN2UxNzgwZiIKICAgIH0sCiAgICAicHJvZHVjdGlvblN1cHBvcnRPd25lckxlYWRlcjEiOiB7CiAgICAgICJlbWFpbCI6ICJzaGFuZS5hLnRoYXRjaGVyQGFleHAuY29tIiwKICAgICAgImd1aWQiOiAiNzdlZGY0ZjUzMzZiNTU3N2E5NTJkMjVkNTRiZDEwZmQiCiAgICB9CiAgfSwKICAicmVnaW9uc1N1cHBvcnRlZCI6IFsKICAgIHsKICAgICAgIklTTzNDb2RlIjogIlVTQSIsCiAgICAgICJuYW1lIjogIlVTIgogICAgfQogIF0sCiAgInJlZ3VsYXRvcnlJbmZvcm1hdGlvbiI6IFsKICAgIHsKICAgICAgInF1ZXN0aW9uQ29kZSI6ICJSUjAyIiwKICAgICAgInZhbHVlcyI6IFsKICAgICAgICAiTm8iCiAgICAgIF0KICAgIH0sCiAgICB7CiAgICAgICJxdWVzdGlvbkNvZGUiOiAiUlIwMyIsCiAgICAgICJ2YWx1ZXMiOiBbCiAgICAgICAgIk4vQSIKICAgICAgXQogICAgfSwKICAgIHsKICAgICAgInF1ZXN0aW9uQ29kZSI6ICJSUjA0IiwKICAgICAgInZhbHVlcyI6IFsKICAgICAgICAiTi9BIgogICAgICBdCiAgICB9LAogICAgewogICAgICAicXVlc3Rpb25Db2RlIjogIlJSMDUiLAogICAgICAidmFsdWVzIjogWwogICAgICAgICJOL0EiCiAgICAgIF0KICAgIH0sCiAgICB7CiAgICAgICJxdWVzdGlvbkNvZGUiOiAiUlIwNiIsCiAgICAgICJ2YWx1ZXMiOiBbCiAgICAgICAgIlllcyIKICAgICAgXQogICAgfSwKICAgIHsKICAgICAgInF1ZXN0aW9uQ29kZSI6ICJSUjA3IiwKICAgICAgInZhbHVlcyI6IFsKICAgICAgICAiWWVzIgogICAgICBdCiAgICB9LAogICAgewogICAgICAicXVlc3Rpb25Db2RlIjogIlJSMDgiLAogICAgICAidmFsdWVzIjogWwogICAgICAgICJZZXMiCiAgICAgIF0KICAgIH0sCiAgICB7CiAgICAgICJxdWVzdGlvbkNvZGUiOiAiUlIwOSIsCiAgICAgICJ2YWx1ZXMiOiBbCiAgICAgICAgIk5vIgogICAgICBdCiAgICB9LAogICAgewogICAgICAicXVlc3Rpb25Db2RlIjogIlJSMTAiLAogICAgICAidmFsdWVzIjogWwogICAgICAgICJObyIKICAgICAgXQogICAgfSwKICAgIHsKICAgICAgInF1ZXN0aW9uQ29kZSI6ICJSUjExIiwKICAgICAgInZhbHVlcyI6IFsKICAgICAgICAiTm8iCiAgICAgIF0KICAgIH0sCiAgICB7CiAgICAgICJxdWVzdGlvbkNvZGUiOiAiUlIxMiIsCiAgICAgICJ2YWx1ZXMiOiBbCiAgICAgICAgIk5vIgogICAgICBdCiAgICB9LAogICAgewogICAgICAicXVlc3Rpb25Db2RlIjogIlJSMTMiLAogICAgICAidmFsdWVzIjogWwogICAgICAgICJQcmUtZGVmaW5lZCBwcm9maWxlcyBiYXNlZCBvbiBqb2IgZnVuY3Rpb24gb3Igcm9sZXMiCiAgICAgIF0KICAgIH0sCiAgICB7CiAgICAgICJxdWVzdGlvbkNvZGUiOiAiUlIxNCIsCiAgICAgICJ2YWx1ZXMiOiBbCiAgICAgICAgIlllcyIKICAgICAgXQogICAgfSwKICAgIHsKICAgICAgInF1ZXN0aW9uQ29kZSI6ICJSUjE3IiwKICAgICAgInZhbHVlcyI6IFsKICAgICAgICAiWWVzIgogICAgICBdCiAgICB9LAogICAgewogICAgICAicXVlc3Rpb25Db2RlIjogIlJSMTgiLAogICAgICAidmFsdWVzIjogWwogICAgICAgICIwIHRvIDMiCiAgICAgIF0KICAgIH0sCiAgICB7CiAgICAgICJxdWVzdGlvbkNvZGUiOiAiUlIxOSIsCiAgICAgICJ2YWx1ZXMiOiBbCiAgICAgICAgIlllcyIKICAgICAgXQogICAgfSwKICAgIHsKICAgICAgInF1ZXN0aW9uQ29kZSI6ICJSUjIwIiwKICAgICAgInZhbHVlcyI6IFsKICAgICAgICAiTGVzcyB0aGFuIDEwMCIKICAgICAgXQogICAgfSwKICAgIHsKICAgICAgInF1ZXN0aW9uQ29kZSI6ICJSUjIxIiwKICAgICAgInZhbHVlcyI6IFsKICAgICAgICAiRW1wbG95ZWVzIgogICAgICBdCiAgICB9LAogICAgewogICAgICAicXVlc3Rpb25Db2RlIjogIlJSMjIiLAogICAgICAidmFsdWVzIjogWwogICAgICAgICJDb2Fyc2UiCiAgICAgIF0KICAgIH0sCiAgICB7CiAgICAgICJxdWVzdGlvbkNvZGUiOiAiUlIyMyIsCiAgICAgICJ2YWx1ZXMiOiBbCiAgICAgICAgIk9uIFRlcm1pbmF0aW9uIgogICAgICBdCiAgICB9LAogICAgewogICAgICAicXVlc3Rpb25Db2RlIjogIlJSMjQiLAogICAgICAidmFsdWVzIjogWwogICAgICAgICJObyIKICAgICAgXQogICAgfSwKICAgIHsKICAgICAgInF1ZXN0aW9uQ29kZSI6ICJSUjI1IiwKICAgICAgInZhbHVlcyI6IFsKICAgICAgICAiTm8iCiAgICAgIF0KICAgIH0sCiAgICB7CiAgICAgICJxdWVzdGlvbkNvZGUiOiAiUlIwMVByaW1hcnlQSUkiLAogICAgICAidmFsdWVzIjogWwogICAgICAgICJOL0EiCiAgICAgIF0KICAgIH0sCiAgICB7CiAgICAgICJxdWVzdGlvbkNvZGUiOiAiUlIwMVNlY29uZGFyeVBJSSIsCiAgICAgICJ2YWx1ZXMiOiBbCiAgICAgICAgIk4vQSIKICAgICAgXQogICAgfSwKICAgIHsKICAgICAgInF1ZXN0aW9uQ29kZSI6ICJSUjAxQWZmaWxpYXRpb25TREUiLAogICAgICAidmFsdWVzIjogWwogICAgICAgICJOL0EiCiAgICAgIF0KICAgIH0sCiAgICB7CiAgICAgICJxdWVzdGlvbkNvZGUiOiAiUlIwMVByaW1hcnkiLAogICAgICAidmFsdWVzIjogWwogICAgICAgICJObyIKICAgICAgXQogICAgfSwKICAgIHsKICAgICAgInF1ZXN0aW9uQ29kZSI6ICJSUjAxU2Vjb25kYXJ5IiwKICAgICAgInZhbHVlcyI6IFsKICAgICAgICAiTm8iCiAgICAgIF0KICAgIH0sCiAgICB7CiAgICAgICJxdWVzdGlvbkNvZGUiOiAiUlIwMUFmZmlsaWF0aW9uIiwKICAgICAgInZhbHVlcyI6IFsKICAgICAgICAiTm8iCiAgICAgIF0KICAgIH0sCiAgICB7CiAgICAgICJxdWVzdGlvbkNvZGUiOiAiUlIwMSIsCiAgICAgICJ2YWx1ZXMiOiBbCiAgICAgICAgIk5vIgogICAgICBdCiAgICB9CiAgXSwKICAicmlzayI6IHsKICAgICJiaWEiOiAiLTEiLAogICAgInJpc2tMZXZlbCI6ICJNZWRpdW0iLAogICAgInJpc2tTY29yZSI6IDEsCiAgICAicnRyIjogLTEKICB9LAogICJzZWN1cml0eUluZm9DbGFzc2lmaWNhdGlvbiI6ICJBWFAgUmVzdHJpY3RlZCIsCiAgInNvdXJjZUNvZGUiOiB0cnVlLAogICJ0ZWNoUGxhdGZvcm1JZCI6IG51bGwsCiAgInRlY2hub2xvZ3lFbnZpcm9ubWVudCI6IFsKICAgICJEaXN0cmlidXRlZCIKICBdLAogICJ2YWxpZGF0aW9uU3RhdHVzIjogIlZhbGlkYXRlZCIKfQ==”,“bucket”:“CentralDB”,“vBucketUuid”:243593261013489}’,
offset: 4389,
partition: 3,
highWaterOffset: 4390,
key:
‘{“schema”:{“type”:“string”,“optional”:false},“payload”:“asset::115217032”}’ }

david.nault · February 7, 2020, 12:40am

Hi Pramod,

I’m glad to hear you’re one step closer to the goal.

If you want the Kafka message to include the Couchbase document as un-encoded JSON, you can configure the connector to use RawJsonSourceHandler (or its cousin RawJsonWithMetadataSourceHandler) as described in the Publishing Raw Json Messages section of the quickstart guide.

Thanks,
David

pramod.k.kaza · February 7, 2020, 2:16am

its working for me , 2 more questions

Is there an option to compress the data while it gets sent ?
Currently eventhough for a particular key , i am only updating 2 out of 10 fileds , the event is sending me the entire document with 10 fileds , including the 2 updated ones … Is there any option to only send 2 fileds as part of the message event ?

david.nault · February 7, 2020, 7:39pm

Hi Pramod,

Is there an option to compress the data while it gets sent ?

The couchbase.compression config property controls whether documents are compressed when sent from Couchbase Server to the connector. It’s enabled by default. The connector always decompresses the document before writing it to the Kafka topic, regardless of this setting.

I haven’t tried this, but apparently Kafka supports compression, which can be enabled either in the Kafka producer client or the broker. I’m not sure if the Kafka Connect framework exposes enough of the client settings to enable compression on the producer (I doubt it, but I might be wrong). Compression can be enabled on the broker when the topic is created; see the compression.type topic configuration option.

2 . Currently event hough for a particular key , i am only updating 2 out of 10 fields , the event is sending me the entire document with 10 fields , including the 2 updated ones … Is there any option to only send 2 fields as part of the message event ?

No, I’m afraid not. This is just how the protocol used by the connector works.

Thanks,
David

pramod.k.kaza · February 7, 2020, 7:51pm

Few more questions from a deployment perspective

We are on Openshift on prem cloud with 3 nodes on primary Data center and 3 nodes on backup data center .
Will i be ending up attaching 1 data connector to each node in primary data center ( total 3)? If "yes " , suppose a CDC event gets trigered from couchbase will all the dataconnectors pointing to 3 nodes be processing the same event and sending 3 messages to the configured queue or only one message gets sent to the configured topic ? is there any specific configuration that i will need to do for this scenario ?

david.nault · February 7, 2020, 8:29pm

Hi Pramod,

One setting you might want to tweak is tasks.max.

If you’re running a single connector worker in standalone mode, that one worker talks to all 3 nodes in the Couchbase cluster, and you can optimize resource usage by setting the tasks.max config property to 1.

If you’re deploying the connector in distributed mode, each distributed worker will handle a subset of the Couchbase partitions (also known as “virtual buckets” or “vbuckets”); each Couchbase change event will be sent to only one of the workers. In this scenario it’s recommended to increase the tasks.max configuration to take advantage of all of the worker nodes. If you want all worker nodes to participate in running the connector, set tasks.max equal to the number of worker nodes.

One connector can only talk to one Couchbase cluster. I’m not sure how to deal with the backup Couchbase cluster; that’s a big architectural question. Our support team might have some ideas, and would be the best people to contact if you want to dive deeper into this subject.

Thanks,
David

patel-saket · April 15, 2020, 3:18pm

Hi Pramod,

I am also getting the same error “org.apache.kafka.connect.errors.ConnectException: Cannot fetch configuration for bucket start_connector_bucket” where start_connector_bucket is my bucket name.

Request Body:-

{
“name”: “abc-3”,
“config”: {
“connector.class”: “com.couchbase.connect.kafka.CouchbaseSourceConnector”,
“tasks.max”: 2,
“connection.cluster_address”: “couchbase-service:8091”,
“connection.bucket”: “start_connector_bucket”,
“topic.name”: “start_connector_topic”,
“connection.username”: “Administrator”,
“connection.passowrd”: “Administrator”,
“dcp.message.converter.class”: “com.couchbase.connect.kafka.handler.source.DefaultSchemaSourceHandler”,
“value.converter”: “org.apache.kafka.connect.storage.StringConverter”,
“transforms”: “ignoreDeletes”,
“couchbase.stream_from” : “BEGINNING”,
“transforms.ignoreDeletes.type”: “com.couchbase.connect.kafka.transform.DropIfNullValue”
}
}

I tried multiple combinations for “connection.cluster_address”: “couchbase-service:8091” but couldn’t get any luck.

Error:-
org.apache.kafka.connect.errors.ConnectException: Cannot fetch configuration for bucket start_connector_bucket
at com.couchbase.connect.kafka.CouchbaseSourceConnector.start(CouchbaseSourceConnector.java:60)
at org.apache.kafka.connect.runtime.WorkerConnector.doStart(WorkerConnector.java:110)
at org.apache.kafka.connect.runtime.WorkerConnector.start(WorkerConnector.java:135)
at org.apache.kafka.connect.runtime.WorkerConnector.transitionTo(WorkerConnector.java:195)
at org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:257)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:1183)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.access$1400(DistributedHerder.java:125)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:1199)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder$15.call(DistributedHerder.java:1195)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

I have deployed all the components (kafka, couchbase and one simple microservice) in openshift under the same namespace.

Is there any additional configuration which you did on couchbase or kafka to fix this issue?
Your inputs will be helpful.

Thanks in advance.

david.nault · April 15, 2020, 4:24pm

Looks like a typo in this property, should be: connection.password

patel-saket · April 15, 2020, 5:30pm

What ??? How did I not notice it ??

Changed it to connection.password and worked like a charm.

Thanks, David