Kafka source: support transactional mode

[sravotto]

CockroachDB changefeeds emit resolved timestamps across all the partitions within a topic. We could leverage it, to allow transactional consistency across multiple rows, by staging events and committed them to the target tables once we get the same resolved timestamp on all the partitions.