feat(kafka): Add Ingestion from Kafka in Ingesters

[cyriltovena]

What this PR does / why we need it:

This adds a new flag to allow starting ingesting from Kafka. It's heavily inspired by Mimir Kafka ingestion.

This is meant to be used as a new set of replica of ingesters.

The idea is simple, we keep the same ingestion from Ingester but allow to ingest from Kafka.
Ingesters now shares partition ownership through the partition ring.

A new downscale partition endpoint is added for downscaling and keeping partition alive until the ingester query window (2h) is passed. The new endpoint is used by the new rollout operator.

Which issue(s) this PR fixes:
Fixes https://github.com/grafana/loki-private/issues/1115

Special notes for your reviewer:

Checklist

• Reviewed the CONTRIBUTING.md guide (required)
• Documentation added
• Tests updated
• Title matches the required conventional commits format, see here
  • Note that Promtail is considered to be feature complete, and future development for logs collection will be in Grafana Alloy. As such, feat PRs are unlikely to be accepted unless a case can be made for the feature actually being a bug fix to existing behavior.
• Changes that require user attention or interaction to upgrade are documented in docs/sources/setup/upgrade/_index.md
• For Helm chart changes bump the Helm chart version in production/helm/loki/Chart.yaml and update production/helm/loki/CHANGELOG.md and production/helm/loki/README.md. Example PR
• If the change is deprecating or removing a configuration option, update the deprecated-config.yaml and deleted-config.yaml files respectively in the tools/deprecated-config-checker directory. Example PR

[owen-d]

LGTM

[owen-d]

Edit2: my train of thought here. tl;dr: no action needed.

if we're moving state to the ingester and persisting via the wal, I think we need to commit after each batch is pulled & accepted, not on a timer. Otherwise, restarting will replay the wal and then process the offsets that's in the wal already, resulting in duplicate lines in storage (if they're accepted to diff chunks, otherwise cut() will dedupe them).

Edit1: looks like you handle this in partitionCommitter.Stop(), which reduces the likelihood a problem here. There's still some gap where duplicates can occur (process dying before it can call the stop() method), but
a) this would be the case between a wal write and subsequent offset commit()
b) running on a timer like this should create fewer offset records in the offsets topic compared to updating them after each write.

All in all, I think this is fine 👍 , considering this problem is going to exist as long as we try to move state from one wal (kafka) to another (ingester) and then update the former. Since we can't make that atomic, this does a good enough job minimizing the blast radius.