Add Backfill Circuit Breaking for Offset Reset Feature

[rseetham]

Currently, this feature always triggers when there's an ingestion delay detected at segment creation. This feature effectively doubles the ingestion which we might not have resources for sometimes. There are no mechanism to stop it if there's any issues or if the cluster is overwhelmed specifically a topic is constantly triggering this feature or if all the topics on a cluster trigger this feature at once. We need some circuit breaking mechanisms.

Proposal

1. Introduce a backfill pause property stream.kafka.consumer.prop.auto.offset.reset.pause. This can be set programmatically (through a controller API that will be added) or through an automatic process details below. This property will be checked before the backfill is triggered at segment creation.
2. Add another property stream.kafka.consumer.prop.auto.offset.reset.maxSegmentsBeforeBackfillSkip - If the number of segments on the table are already high, we will not trigger backfill. In case ingestion is permanently high and not spiky, the ingestion doubling is causing more segments to be created leading to znode limit being reached faster.
3. Add a dsc property maxConcurrentBackfillsPerController. Per controller if there's more than this number of backfills ongoing, we will not trigger any more backfills. In the case where the cluster was down and restarted after a while, all topics will have ingestion lag. This means backfill will trigger for all tables. This is not really necessary in the case because cluster can self recover. Instead if all topics backfill, we will create a lot of topics for this. Since topics cannot be cleanly removed, we will need to carry these topics in the stream config forever which is not ideal.
4. If a backfill is ongoing, we will not trigger another backfill until the current backfill is complete. We will also pause the backfill using the property in point 1. and emit a metric in this case. This featurre was built for occasional spikes. If backfills are constantly triggering then there's more throughput than expected and we should be increasing the number of partitions on the main topic.

[rseetham]

@lnbest0707 Could you take a look?

[chenboat]

1. and 2) looks good to me. 3 is an internal feature (DSC config) which Pinot open source does not have. Should we introduce a new controller config?

2. means no concurrent backills on the same table, right?

[rseetham]

2. Is a backfill pause property on the table config at a topic level that's auto updated.
3. Does not exist even internally. I'll add it as a controller property to OSS.
4. Is to prevent simultaneous backfills at a topic level.