Sources

[purplefox]

Implement sources for ingesting data from Kafka.

Much of the plumbing is already in place for this.

One instance of a particular source will be deployed on each node of the cluster. The source instance will maintain a set of Kafka consumers (can be just one for phase 1). As a batch of records is fetched from the consumer it should be passed to the source.ingestRows method for processing. If the batch is processed without error the consumer offsets for the batch should be committed.

If a source has a key already that should be used for the key in the source table. If not, then the user can optionally specify a key from an existing field (e.g. timestamp). A monotonically increasing value can also be generated for the key if needed.