Support consumer de-aggregation in Kinesis

Kinesis supports producing aggregated (aka batched) record. Thus, kinesis consumer also has support for de-aggregating the records. Refer - https://docs.aws.amazon.com/streams/latest/dev/kinesis-kpl-consumer-deaggregation.html

Kinesis provides an `AggregatorUtil` (https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/retrieval/AggregatorUtil.java) that can be used in the `KinesisConsumer` implementation. An example usage of this util can be found in the beam repo (https://github.com/apache/beam/blob/master/sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/kinesis/SimplifiedKinesisClient.java#L269)

Even though the records in the batch have the same sequence number, we can append the sub-sequence number to Pinot kinesis' `StreamMessageOffset`. Changes should be fairly trivial to do this.

Labels: `enhancement` , `kinesis`, `help-wanted`

Start by locating the KinesisConsumer implementation and its StreamMessageOffset handling. Review AWS's AggregatorUtil and the Apache Beam SimplifiedKinesisClient example linked in the issue, then verify that aggregated records are de-aggregated and each sub-record receives an offset containing its sub-sequence number.

Written by the indexing model from the issue text.