add support to handle records with null value

[bionarkos]

tombstone records currently cause the sink-task to stop, as an exception is thrown. Added a new config property "behavior.on.null.values" to control this behavior in a more granular fashion. The following enum values are allowed:

• ignore (default)
• fail

• log

  Also added details (partition,offset) about which record is failing on the given topic.

[jeqo]

@bionarkos thanks for the PR! Though I would like to validate first if this feature is actually needed.

Have you considered using transforms (a.k.a SMT) to filter1 tombstones2?

For instance, by adding:

{
    //...
    "transforms": "filter",
    "transforms.filter.type": "org.apache.kafka.connect.transforms.Filter",
    "transforms.filter.predicate": "isTombstone",
    "predicates": "isTombstone",
    "predicates.isTombstone.type": "org.apache.kafka.connect.transforms.predicates.RecordIsTo
mbstone"
}

you may get the same behavior without changes on the connector

[bionarkos]

Thank you very much @jeqo for the good tip, the filtering possibility solved my issue, so I can close this PR. :bow: