for example password fields must be covered with *** mask

We need to create a Contributing guideline to this project.

It will enable a person to understand what it requires to contribute to this project.

Feature request related to a problem

• When using ephemeral bucket type (Metadata bucket also same) connector get started successfully but mutation events does not listen by the connector. Also there is no error logs everything seems OK. Then trigger a mutation from couchbase and run project on my local, there is no events received from dcp. We though it might be DCP side but we tested it via couchbase eventing function to receive mutation/delete logs same bucket and collection it works.
  issue link

Describe the solution you'd like
dcp-client observes and desires data to persist on Couchbase. Persistence is not required on Ephemeral Bucket. When adding following config to your config.yaml you can pass that logic.

rollbackMitigation:
  disabled: true

Describe alternatives you've considered
Adding new feature to detect bucket type and apply the config as I mentioned above automatically.

After the cluster configuration changed, lag metrics continuously increased.

This behavior causes false alerting behaviour! We need to debug and solve it asap!

all other connectors using 1.20 without problem. so we can upgrade core library.

like kafka auto.offset.reset: latest
currently our stream working like kafka auto.offset.reset: earliest

currently we are setting nil

we need to implement runtime javascript execution on mapper with eventing features and contract.

In current implementation, I noticed healthcheck function ¹ does not support for retrying along with graceful handling. ²

Proposal
It would be nice to add a something failureThreshold to ConfigHealthCheck. So that after ping fails failureThreshold times in a row, go-dcp-client should treats the CB connection as unhealthy and triggers a restart/reconnect for that specific server.

Happy to discuss reasonable.

for reindex batch jobs, it would be good to have such a feature.

The Problem
When running a DCP app in a Multi-AZ setup. We can only run as active-passive because the same messages will be processed in multiple availability zones, and messages will be processed as duplicated.
In an active-passive setup, when we need to failover our dcp application, we can not continue where the previous dcp app instance left off. This is because sequence numbers of mutations are changed in the target couchbase cluster after mutations are replicated with XDCR.

Proposed Solution
Our proposed solution for a failover scenario is to start the dcp app with snapshot mode as it listens to changes from the beginning. In this case, we need a configuration field such as SkipUntil: "2024-08-12T13:40:03+0000" that the app will skip changes before this date.

A possible alternative solution
If we could find virtual buckets' corresponding sequence numbers in the target cluster, we could find a way to create new offset storage records for virtual buckets. However, we couldn't find a way to do this yet.

Currently, we can only set metadataBucket but we also want to configure metadataScope and metadataCollection. It uses default values for them.

We could use a standard library for writing test assertions. For example,

instead of assert.Equal -> we can use reflect.DeepEqual

for example, we need to detect magma storage and validate/apply change streams config.
dcp.config.enableChangeStreams

Describe the bug
When more than one couchbase data node IP is given, an error is received when connecting because the default port is not set.

Expected behavior
The default port can be 8091.

we need to expose membership and rebalance metrics to prometheus endpoint.

example metrics:

total member
rebalance count
rebalance duration? 
membership type
...

replica changes
absent nodes
new nodes
...

Is your feature request related to a problem? Please describe.
Currently, we can't create multiple connectors that api's enabled, because each connector tries to create a new api. We need to support getting the HTTP server from outside.

Also, we need to find a way to show the metrics of each connector separately.

including Trendyol/go-dcp-couchbase#18

Is your feature request related to a problem? Please describe.
I need to sync a CB table to another database (may not be CB) or micro service. Current go-dcp library is able to capture the latest changes to the table. However, my project also needs the initial snapshot of the table because not all the docs are updated actively.

Describe the solution you'd like
It would be ideal to allow a go-dcp client (connector) to capture the initial snapshot of a given table before starting reading the change logs. The snapshot events may be directly read by the connector and published to kafka without being saved to the change logs. After all the snapshot events are processed, the connector then starts reading the change logs from where the snapshot ends. The behavior could be similar to the Debezium MySQL connector. However, implementing the same for CouchBase is technically a lot more challenging.

Describe alternatives you've considered
There is no clear alternative. It's hard to implement a separate snapshot tool that is seamlessly integrated with a dcp connector.

Additional context
The open-source Couchbase source connector used to have a configuration option "use_snapshot" but dropped in recent releases because it didn't work well.
Ref this discussion for more details: https://www.couchbase.com/forums/t/couchbase-kafka-connector-connection-error-cannot-fetch-configuration-for-bucket/35047

In Dcp stream make mainListenerChannel buffered channel and make buffer size configurable.

We can use ConfigApi to enable admin and metric endpoints

The current implementation of the go-dcp-client library does not allow configuration of the couchbase connection timeout ¹. However, it would be beneficial to make this configurable to allow for greater flexibility in handling connections.

we need to fetch server version to set these configurations

dcp.config.disableExpiryOpcode
dcp.config.disableStreamEndByClient

currently we are expecting from user to set them because gocbcore did not provide server version.

Describe the bug
We are getting panic: send on closed channel exception while closing the application.

Screenshots

Version:

• OS: macOS
• Golang version: v1.20
• github.com/Trendyol/go-kafka-connect-couchbase v0.0.43

Is your feature request related to a problem? Please describe.
The commit function writes 1024 vbucket checkpoint documents to Couchbase. There may be a small number of transactions but either way, we make 1024 write operations.

Describe the solution you'd like
Update vbucket checkpoint documents that were updated after the last commit.

Is your feature request related to a problem? Please describe.
Checkpoint logs almost fill 2.5m of data in a day.

Describe the solution you'd like
We should define the severity of all the logs and clients must be able to configure via configuration.

Currently, we are getting Couchbase's host from the config file as

hosts:
  - xx.xxx.xx.xxx:8091
  - xx.xxx.xx.xxx:8091
  - xx.xxx.xx.xxx:8091

We need to support service discovery URLs like couchbase://hostname.external

This feature is already implemented in gocb, we can get the related codes and implement ours like this

my client side solution;

var clientRecordKey = "_txn:client-record"

func mapper(event dcpescb.Event) []dcpescbd.ESActionDocument {
	if string(event.Key) == clientRecordKey {
		return nil // ignored for mapper because _txn:client-record metadata document created in each collection in the cluster involved with transaction metadata.
	}
}

https://docs.couchbase.com/java-sdk/current/concept-docs/transactions-cleanup.html

We need to check if dcp works or not via an endpoint.

If go-dcp-client provides an endpoint like "/_healthcheck" and it returns the status of the dcp client would be great.

We may need to start stream from latest instead of beginning, we need to have a config to adjust this.

https://github.com/Trendyol/go-dcp/blob/master/example/config_k8s_stateful_set.yml#L11 while using statefulset membership, if we want to scale up our application we need to make a deployment(to update config file). If we set totalMembers as an environment variable we can update the totalMembers without any deployment.