Support index mapping strategy

[nizanifrach]

Problem

Currently, there is no way to configure an index strategy to map a custom index name by topic and message payload. To insert a document to a custom index name at the moment, the developer needs to create a specific topic name that corresponds with the Elasticsearch index name

#595 #385 #341

Solution

In this solution, we introduce a new configuration index.mapper which gets a classpath as an argument of a custom implementation of the IndexMapper.java interface. the getIndex() function in the interface return the custom index name.

For backward compatibility reasons, the index name creation logic was transferred to a DefaultIndexMapper.java which will be loaded if no other configuration is specified.

Does this solution apply anywhere else?

• [x] yes
• [ ] no

If yes, where?

I took inspiration from the MongoDB source connector https://github.com/mongodb/mongo-kafka/tree/master/src/main/java/com/mongodb/kafka/connect/source

Test Strategy

Unit tests:

• JsonNode converter tests.
• IndexMapper configuration test.
• IndexMapper logic tests.

Integration tests:

• All integration tests pass, keeping backward compatibility.

tested manually as well.

Testing done:

• [x] Unit tests
• [ ] Integration tests
• [ ] System tests
• [x] Manual tests

Release Plan

This feature is backwards compatible. Merging to master

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[nizanifrach]

Hi @snehashisp @arihant-confluent @aniketshrimal , Your review and input on this feature would be highly appreciated. Thank you 🙏