Doesn't allow writing duplicates in bulk writer and corresponding tests

tvaron3 commented 1 year ago

Type of PR

[ ] Documentation changes
[x] Code changes
[x] Test changes
[ ] CI-CD changes
[ ] GitHub Template changes

Purpose of PR

The bulk upsert operation will sometimes write data with duplicate id and partition key incorrectly. This change prevents duplicate items to be sent to the bulk upsert operation. It should only send the latest item. The feature is hidden behind a new config that is set to true by default. The flag added is called "connect.cosmos.sink.bulk.compression.enabled".

Observability + Testing

What changes or considerations did you make in relation to observability?
Did you add testing to account for any new or changed work? I added a unit test and a integration test for the changes I made in the bulkwriter.

Review notes

Issues Closed or Referenced

Closes # (this will automatically close the issue when the PR closes)
References # (this references the issue but does not close with PR)

TheovanKraay commented 1 year ago

LGTM - nit: can we please add the actual flag in the PR description?

TheovanKraay commented 1 year ago

LGTM - nit: can we please add the actual flag in the PR description?

I saw you updated, but I meant show the value that customer will need to update in their config file for the connector - I think its connect.cosmos.sink.bulk.duplicates.enabled?

tvaron3 commented 1 year ago

Theo I changed it again and the value they would need to change is "connect.cosmos.sink.bulk.compression.enabled".

microsoft / kafka-connect-cosmosdb