Go implementation of the Kafka Connect Couchbase.

Go Dcp Kafka streams documents from Couchbase Database Change Protocol (DCP) and publishes Kafka events in near real-time.

• Less resource usage and higher throughput(see Benchmarks).
• Custom Kafka key and headers implementation(see Example).
• Sending multiple Kafka events for a DCP event(see Example).
• Handling different DCP events such as expiration, deletion and mutation(see Example).
• Kafka compression support(Gzip, Snappy, Lz4, Zstd).
• Kafka producer acknowledges support(fire-and-forget, wait for the leader, wait for the full ISR).
• Metadata can be saved to Couchbase or Kafka.
• Managing batch configurations such as maximum batch size, batch bytes, batch ticker durations.
• Scale up and down by custom membership algorithms(Couchbase, KubernetesHa, Kubernetes StatefulSet or Static, see examples).
• Easily manageable configurations.

The benchmark was made with the 1,001,006 Couchbase document, because it is possible to more clearly observe the difference in the batch structure between the two packages. Default configurations for Java Kafka Connect Couchbase used for both connectors.

Struct Config

func mapper(event couchbase.Event) []message.KafkaMessage {
	// return nil if you wish to discard the event
	return []message.KafkaMessage{
		{
			Headers: nil,
			Key:     event.Key,
			Value:   event.Value,
		},
	}
}

func main() {
	c, err := dcpkafka.NewConnector(&config.Connector{
		Dcp: dcpConfig.Dcp{
			Hosts:      []string{"localhost:8091"},
			Username:   "user",
			Password:   "password",
			BucketName: "dcp-test",
			Dcp: dcpConfig.ExternalDcp{
				Group: dcpConfig.DCPGroup{
					Name: "groupName",
					Membership: dcpConfig.DCPGroupMembership{
						RebalanceDelay: 3 * time.Second,
					},
				},
			},
			Metadata: dcpConfig.Metadata{
				Config: map[string]string{
					"bucket":     "checkpoint-bucket-name",
					"scope":      "_default",
					"collection": "_default",
				},
				Type: "couchbase",
			},
			Debug: true},
		Kafka: config.Kafka{
			CollectionTopicMapping: map[string]string{"_default": "topic"},
			Brokers:                []string{"localhost:9092"},
		},
	}, mapper)
	if err != nil {
		panic(err)
	}

	defer c.Close()
	c.Start()
}

File Config

File Config

Check out on go-dcp

You can also use all DCP-related metrics explained here. All DCP-related metrics are automatically injected. It means you don't need to do anything.

Go Dcp Kafka is always open for direct contributions. For more information please check our Contribution Guideline document.

Released under the MIT License.