CDC inserts 500k duplicates into Kafka

[yugabyte-ci]

Jira Link: DB-7993

[vaibhav-yb]

Update

Context

GetTabletListToPollForCDC

We have an API to get the checkpoint for tablets i.e. GetTabletListToPollForCDC which gets the checkpoint from the cdc_state table in service. We can also request the children of a given parent tablet once it is split and we will receive the children tablets along with their checkpoint values.

Now assume we have a tablet T and it got split into t1 and t2 - so once the client will ask for the children tablets, we will get t1 and t2 along with their checkpoint values. Now while retrieving the checkpoint, we copy the parent tablet's checkpoint to the children tablets but in case the parent itself is not found in the state table, service returns as a checkpoint with OpId as 0.0.

Invalid checkpoint

If a tablet is not polled even once, service returns an invalid checkpoint with OpId as -1.-1.

Reason for duplicates

1. Stream was created on a namespace
2. A table test_streaming is being polled - there are other tables (test_2, test 3) in the namespace at the moment as well, they are not being polled anywhere for changes
3. Table test_2 is dropped
4. Stream ID's state is marked as DELETING_METADATA
5. Simultaneously, one of the tablets for table test_streaming gets split
6. A background thread runs which looks up for the tablet+stream ID combination having state as DELETING_METADATA and deletes the tablet
7. Client detects the split around the same time and asks for children - now since the parent tablet is deleted, the checkpoint for the children will be sent as 0.0 - and when GetChanges will be called on this checkpoint, an INTERNAL_ERROR will be thrown with a message that the checkpoint in the state table is invalid. This is because the children tablets haven't been polled before.
8. At the same time, the client restarts and it again asks the checkpoint for all the tablets including children tablets from the service - service sends back the checkpoint as invalid i.e. -1.-1 - this checkpoint is then used to evaluate whether or not to take the snapshot. If it is invalid then the client takes the snapshot.
9. Since snapshot is taken for the children tablets, this causes duplicates in Kafka.

[EDIT]

The fix was merged with the ticket as #19348