feat: Enable tracking of vBucket stream consumption by providing a callback triggered on each DCP event

[ziollek]

Why is this beneficial?

Consider an application that exposes an API and:
a. Maintains in-memory aggregations based on data stored in DCP.
b. Needs to determine when it has caught up with historical data before serving requests.

If we use DCP to feed the application with both historical and ongoing data, requirement (b) presents a challenge.

It is possible to fetch the current sequence numbers at startup using dcp.GetClient().GetVBucketSeqNos() and then start DCP. However, not all events will be passed to our listener due to some being skipped in stream.listen.

Proposed solution

The solution involves replacing the listener callback in the dcp struct with an interface that defines two methods:

• ConsumeEvent(ctx *ListenerContext), which maintains compatibility with the existing Listener callback type.
• TrackOffset(vbID uint16, offset *Offset), which will be invoked within stream.setOffset.

type Consumer interface {
	ConsumeEvent(ctx *ListenerContext)
	TrackOffset(vbID uint16, offset *Offset)
}

Backward Compatibility

To maintain backward compatibility, the existing factory method:

func NewDcp(cfg any, listener models.Listener) (Dcp, error)

will be preserved.

Additionally, a new factory method will be introduced to support the extended interface:

func NewExtendedDcp(cfg any, consumer models.Consumer) (Dcp, error)

Alternative Approaches

• Modifying event-passing logic: Adjusting how events are passed to the listener to provide more control over which events should be consumed and to determine offsets within the listener callback.
  • Downside: This approach significantly impacts the streaming behavior and poses challenges for backward compatibility.
• Fetching data from Couchbase during the warm-up stage: Retrieving the necessary data using alternative mechanisms (e.g., SQL++) before starting the DCP stream.
  • Downside: This adds complexity to an otherwise straightforward approach that relies solely on go-dcp.

[erayarslan]

hi @ziollek
thank you so much for your feature request and contributions! 🚀
i think its awesome because we also need this kind of flexibility too to understand for example when our reindex job catchup to current system.
i will review this asap.
btw did you consider to implement this feature on connectors?
below are our current connectors using go-dcp;

go-dcp-elasticsearch
go-dcp-kafka
go-dcp-couchbase
go-dcp-sql

[ziollek]

Oh, I didn't know about these on-top projects :). Currently, I need to feed application memory from the DCP stream, but I will take a closer look in my spare time.
I've prepared changes in my fork it is quite a straightforward adjustment.

[erayarslan]

i open pr with your fork #125 lets go with that 🚀