feat: Implement WAL segment ingestion via Kafka with partition ring

[cyriltovena]

• Set up Kafka consumer for WAL segments
• Implement partition ring for managing assignments
• Add delayed downscale endpoint for graceful partition consumption
• Implement 15s block building and flushing to metastore/object storage
• First pass on the encoding/decoding of push request to Kafka

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

[grobinson-grafana]

We should add a check for input size so we don't write empty blocks.

 // Flush writes the accumulated data to storage and updates the metadata store
 func (c *consumer) Flush(ctx context.Context) error {
+       if c.writer.InputSize() == 0 {
+               return nil
+       }
+
        start := time.Now()
        c.metrics.flushesTotal.Add(1)
        defer func() { c.metrics.flushDuration.Observe(time.Since(start).Seconds()) }()