[Proposal] Pubsub Indexing Service

### Motivation

For streaming ingestion, Kafka/Kinesis queues are the two primary options for Druid as of now.
Proposal is to write an extension that will allow Data ingestion from Google Cloud Pubsub.

### Proposed changes

The proposed extension will work in a manner very similar to Seekable Stream Supervisors & Tasks that we currently have & will be a simplified version of the same in most cases.

Key differences between pubsub & kafka queues in context of this implementation are as follows,
1. Unlike Kafka, PubSub does not have a concept of ordered logs. Packets are pulled in batches from the cloud & acknowledgements are sent after successful processing per packet.
2. While Kafka has a Topic/Partition hierarchy such that one packet for a topic goes only in one of it's partitions, PubSub has Topic/Subscription hierarchy where any packets that is sent to a topic is replicated in all of the subscriptions. In the ideal case, each packet should be pulled only once from a subscription.

One key design decision that we are suggesting is to have a completely independent extension that does not share any logic with the Seekable Stream Ingestion extensions.

1. The ingestion specs will be very similar to Kafka Ingestion specs, with configuration options which mostly overlap but with some additions & removals.
2. The structure & patterns in the code will be a simplified merger of SeekableStreamIndexingService & KafkaIndexingService.

One key challenge to note is that PubSub does not provide exactly-once semantics like Kafka.
This means that in cases of high lag or failures, consumers may pull the same packet more than once from the subscription.
One reasonable approach to tackle this is best effort deduplication.
There are techniques to minimize duplication, a few of them explained as follows,

- Ack deadline tweaks: A consumer can reset ack deadlines when some packets are taking more time to process than expected. Also, having a sensible ack deadline configuration will also have a massive impact to start with.
- Configurable LRU Cache / Bloom filter based dedup: A local sketch of unique packets encountered so far in the previous 'x' minutes can prevent duplicate insertion on the basis of unique packet ids.

It could be possible to provide perfect deduplication using a shared key-value store but that would be out of scope for the first version of the extension.

Additional details are as follows,

**At-least once semantics**: The tasks will send acknowledgements to pubsub only when batch persist succeeds. Thus, in case of failures the packets will be re-sent from the cloud & at-least once semantics will be guraranteed.

**Basic description of the algorithms**: The extension consists of supervisors & tasks, which are explained as follows,

- Supervisor runs in an infinite loop which handles creation and management of tasks If more tasks are required to maintain active task count, it submits new tasks & constantly polls the state of previously started tasks. Supervisors itself can be started & stopped.

- A single task would pull in a batch from pubsub (relevant tuning parameters should be available in config), hands off the packets for persistence & sends back an acknowledgement messages to pubsub for the batch on successful persist.

**Linear scalability**: The system can be scaled just by adding more tasks distributed to whatever druid nodes are discovered. Since multiple consumer instances can pull from the same subscription, we can distribute as many tasks on different machines as we require to keep the lag low.

**Failure handling**: Failure scenarios & corresponding handling approaches are as follows,

- If a task crashes before data is successfully persisted, it should be ignored as supervisor will ensure that a new task will take it's place & pubsub would redeliver the messages.
- If data is successfully persisted, but acknowledgement to pubsub fails, we would attempt basic retry with exponential back-off strategy with some time cap.
- If supervisor crashes due to some irrecoverable machine failure, it should not have any impact on the tasks that have been previously submitted. This however will halt creation of any new tasks.

### Rationale

There were 2 approaches that were under consideration for this feature.

1) Consume from Pubsub in a manner similar to Kafka.
https://github.com/apache/druid/pull/9116

2) Consume from Apache Beam (GCP dataflow) which can pull from pubsub & push to Druid. Since this will be a push based approach while Druid works much more nicely with pull based ingestion, it was discarded for now.
This approach opens up a lot of possibilities but at the cost of having intermediate workers running Apache Beam pipelines.

### Operational impact

Since the extension is a completely new feature with different pathways, it does not have operational impact on running druid clusters.
The dependencies that are globally being updated are the Guava & Guice versions which are very outdated & do not work with latest pubsub libraries. Any possible regression from it should be caught by automated tests.

### Test plan
TODO

### Future work

Configurable options for perfect deduplication strategies are among the possibilities for future improvements.

Start by reading the existing SeekableStreamIndexingService and KafkaIndexingService patterns, then review the Kafka ingestion specs described in the proposal. Define the extension boundaries, supervisor and task behavior, acknowledgement and failure handling, and a test plan for the at-least-once semantics. Done means a reviewed design and implementation scope for Google Cloud Pub/Sub ingestion, including automated coverage.

Written by the indexing model from the issue text.