Sink connector ID configuration

[ricellis]

Currently the sink connector handles _id as follows:

replication option	_id present in SinkRecord	_id written
true	true	_id
true	false	none, new _id generated by Cloudant
false	true	<topic-name>_<partition>_<offset>_<_id>
false	false	none, new _id generated by Cloudant

I don't think the _id should be decided by a seemingly unrelated configuration option.

We should instead document the use of SMTs to faciliate customization of the _id; namely:

• ReplaceField
  • Rename a current field to _id
  • Exclude the _id field to force Cloudant to generate

We should provide (and document) a new SMT class (KeyToDocId ?) that can insert the message key into the value _id field (noting that it should be a String schema or stringifiable).

In the case that the required _id field is not present or the record has a null key then we should not pass an ID to Cloudant and just let it generate a UUID. In the case that the user wanted all _id generated they could use ReplaceField with an exclude to remove any existing _id and the default id mode.

We should also document that it is possible to use further SMTs to customize e.g.

• ValueToKey and ExtractField to convert some field to the key and then using the KeyToDocId transform to use the new key as an ID.

[ricellis]

Rather than a sink connector configuration option I think providing (and documenting) a SMT that is able to insert the message key into the message value as an _id field is the way to go.

[ricellis]

Looking at the complexity that is coming out in #82 from trying to handle all the variations of inserting an _id into different types of messages and associated schema updates I think we need to re-think how we handle this.
One option would be to:

1. Modify the ConnectRecordMapper to check for the presence of a custom header on the record and use the value of that header as the ID.
2. Use the org.apache.kafka.connect.transforms.HeaderFrom to convert a key to the header, avoiding the need for us to have any custom SMT.

For the header, we would ensure the StringConverter was used as we only want strings for IDs. We would need to document the name of the header and the expectation that the value is a string. (this would cause errors for things that cannot convert cleanly to strings, inline with general Kafka behaviour on converters).

We would overwrite existing _id in the JSON blob before writing to Cloudant (with a warning).