Enable conditional writes in connector

[subhashiyer9]

Problem

Solution

• Adds a config to enable conditional writes, the default being true
• The conditional write will be enabled only when scheduled interval is set and store kafka key/headers config is disabled to retain the existing delivery semantics
• On a conditional write exception, a FileExistsException is rethrown , which being a retriable exception will reset the offset and poll again accordingly
• In case of FileExistsException, connector will scan the s3 bucket for next available file which is not present in S3 by making consecutive get object requests. If the get api fails with 403 error, it will simply increment to next offset.
• To enable determining the next available file, the equivalent target file name is cached for initial few records, had the record been the first record in the file. The offset will be reset to the one whose equivalent file does not exist in S3
• Use LinkedHashMap to store commitFiles instead of existing HashMap to ensure files with smaller start offsets (i.e. the ones which were added to the map first) are written to S3 first. This is required to ensure we safely increment to next offset in case of conflict

Does this solution apply anywhere else?

• yes
• no

If yes, where?

Test Strategy

Testing done:

• Unit tests
• Integration tests
• System tests
• Manual tests
• Performance tests - TODO
• Chaos tests - TODO

Release Plan

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}