Support setting a custom deletewritemode.strategy

[ArneKlein]

Hi,

we currently have the scenario where we have to create multi version of the same document based on a key in the value and delete all versions based on the key once we receive a tombstone event.

To support this scenario, it would be great to enable setting a custom "deletewritemodel.strategy" just as we currently can for the normal writemodel.strategy

https://jira.mongodb.org/browse/KAFKA-320

[ArneKlein]

Apologies for the delay in reviewing.

I've added a couple of minor nits for being included into master. I would like to see some tests added to test the new logic

Thank you for the review, I added basically the same tests as for getWriteModelStrategy(). Please let me know if this is sufficient

[rozza]

LGTM

[ArneKlein]

Any chance that this can get merged?

[jyemin]

Note: the latest patch doesn't compile:

com/mongodb/kafka/connect/sink/MongoSinkConfigTest.java:812: error: incompatible types: Optional<WriteModelStrategy> cannot be converted to WriteModelStrategy
               cfg.getMongoSinkTopicConfig(TEST_TOPIC).getDeleteWriteModelStrategy();