dmaier-couchbase

{"schema":{"type":"struct","fields":[{"type":"string","optional":false,"field":"event"},{"type":"int16","optional":false,"field":"partition"},{"type":"string","optional":false,"field":"key"},{"type":"int64","optional":false,"field":"cas"},{"type":"int64","optional":false,"field":"bySeqno"},{"type":"int64","optional":false,"field":"revSeqno"},{"type":"int32","optional":true,"field":"expiration"},{"type":"int32","optional":true,"field":"flags"},{"type":"int32","optional":true,"field":"lockTime"},{"type":"bytes","optional":true,"field":"content"}],"optional":false,"name":"com.couchbase.DcpMessage"},"payload":{"event":"mutation","partition":417,"key":"vehicle::a6c9d5d6-3735-4a87-bd2d-e79dd7b38b89","cas":1485859054995963904,"bySeqno":23,"revSeqno":1,"expiration":0,"flags":33554432,"lockTime":0,"content":"eyJ0eXBlIiA6ICJ2ZWhpY2xlIiwiaWQiIDogImE2YzlkNWQ2LTM3MzUtNGE4Ny1iZDJkLWU3OWRkN2IzOGI4OSIsInZfdHlwZSIgOiAiQlVTIiwic3BlZWQiIDogNTAsImZ1ZWxfbGV2ZWwiIDogODMsInJvdXRlIiA6IHsidHlwZSIgOiAicm91dGUiLCJmcm9tIiA6IHsidHlwZSIgO

dmaier-couchbase

$KAFKA_HOME/bin/kafka-console-consumer.sh --new-consumer --bootstrap-server localhost:9092 --topic test-cb --from-beginning

dmaier-couchbase

[2017-04-06 14:52:10,117] INFO Finished starting connectors and tasks (org.apache.kafka.connect.runtime.distributed.DistributedHerder:829)
[2017-04-06 14:52:10,119] INFO CouchbaseSourceTaskConfig values: 
connection.bucket = vehicles
connection.cluster_address = [ubuntu-local]
connection.password = [hidden]
connection.ssl.enabled = false
connection.ssl.keystore.location = 
connection.ssl.keystore.password = [hidden]
connection.timeout.ms = 2000
dcp.message.converter.class = com.couchbase.connect.kafka.converter.SchemaConverter

dmaier-couchbase

$KAFKA_HOME/bin/connect-distributed.sh $KAFKA_HOME/config/couchbase-distributed.properties

dmaier-couchbase

{
  "name" : "test-couchbase",
  "config" : {
      "connector.class" :  "com.couchbase.connect.kafka.CouchbaseSourceConnector",
      "connection.cluster_address" : "ubuntu-local",
      "connection.bucket" : "vehicles",
      "connection.password" : "test",
      "connection.timeout.ms" : "2000",
      "tasks.max" : 2,
      "topic.name" : "test-cb",

dmaier-couchbase

curl -XPOST http://localhost:8083/connectors -d "@config/couchbase-distributed.json" -H "Content-Type: application/json"

dmaier-couchbase

## Connect
bootstrap.servers=localhost:9092
group.id=connect-cluster
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status

## Converters
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

dmaier-couchbase

$KAFKA_HOME/bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test-cb
Topic:test-cb PartitionCount:3 ReplicationFactor:3 Configs:
Topic: test-cb Partition: 0 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
Topic: test-cb Partition: 1 Leader: 2 Replicas: 2,0,1 Isr: 2,0,1
Topic: test-cb Partition: 2 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2

dmaier-couchbase

$KAFKA_HOME/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic test-cb

dmaier-couchbase

cd $HOME/Downloads/kafka-connect-couchbase-3.1-2.0
cp etc/kafka-connect-couchbase/*.properties $KAFKA_HOME/config/
cp share/java/kafka-connect-couchbase/*.jar $KAFKA_HOME/libs