connectors: Add Kinesis Source and Sink

[jacksonrnewhouse]

This adds support for Kinesis via a source and sink. Sources and Sinks are configured via the name of the stream.
Source also takes a "source.offset" of either "Earliest" or "Latest", similar to kafka.

Kinesis Source

Execution Flow

The source operates by having an active future for every open shard the operator owns. These futures are all advanced via a FuturesUnordered, and the futures are all just labeled bits of compute, without any data that should be owned by the overall control loop. This lets us take accurate snapshots without waiting for the futures to complete. Unfortunately, the futures produced by calling something like async fn my_async_function(&self) are always owned by self. The BoxedFuture gets around this, at the cost of some complexity.

In order to find the affiliated shard state, the futures are returned with a name, which corresponds to the shard_id.

This approach is similar to what I implemented for the Filesystem Sink. Might be worth figuring out how to better structure it or if there's a replacement we prefer.

Semantics

The operator consumes each shard of data in order, similar to the Kafka source. The set of shards that a subtask is responsible for is determined solely by the hash of the shard_id. Flink defaults to this behavior, but also has an option to evenly divide the hash space.

There are no order guarantees for how data is read off of different shards. While we eagerly fetch from all live shards, it is possible that one might fall behind. Since watermarks are at the subtask level, it is possible that a shard that falls behind will have its data dropped. Finally, Kinesis has the notion of "parent" and "child" shards, with the child shard having data after the parents. Ensuring this while letting child shards be on different subtasks is not currently possible, and no effort was made to ensure this.

Kinesis Sink

Execution Flow

The Kinesis sink operates in batches. Right now it waits until one of the following conditions: 500 messages, 4.5MB of data, or 1s has passed. The first two are dictated by the constraints on the PutRecords method, while the timeout is there to ensure we don't have messages linger in the sink.

The PutRecords method allows for partial success of writes, often because of capacity limits on some but not all of the shards in the stream. Once we decide to flush we currently try repeatedly to finish the batch. While this is happening no new messages are consumed, leading to back-pressure.

Semantics

Because the batch puts can succeed only some of the records, the output order is not guaranteed. Additionally, because Kinesis requires a key, if there isn't one in the Pipeline DAG we add a UUID4.

Right now if there's a key over 256 bytes we won't be able to write. I think our SQL solution will, in general, use the UUID4, but when we add partitioning support, we should have a plan for this.

[mwylde]

Initial review of just the source operator. I'll spent some more time later today trying to fully understand the control flow there, and digging into the sink.