…for comprehension

During a graceful shutdown the partition stream is ended. The runloop however, is still happily fetching records for that stream. These are then put in the stream's queue even though they will never be read. This is taking away network and CPU at a moment where the application should focus on doing a clean shutdown quickly. It could even cause an OOM for application that are tuned for the case where processing happens almost immediately.

With this change, when a stream is ended:
- the partition is paused,
- already fetched new records are not added to the stream's queue.

Also: less logging at start of a poll.

…rruption of withStream

…-control

…-control

…-control

By using ZIO.async, we no longer need a reference to the zio runtime,
nor do we need the `exec` trickery anymore.

Also: fix typo and make metric descriptions consistent.

When many hundreds of partitions need to be consumed, an excessive
amount of heap can be used for pre-fetching. The
`ManyPartitionsQueueSizeBasedFetchStrategy` works similarly as the
default `QueueSizeBasedFetchStrategy` but limits total memory usage.

Refactoring of the producer so that it handles errors per record.

Zio-kafka applications always pre-fetch data so that user streams can
process the data asynchronously. This is not compatible with auto
commit. When auto commit is enabled, the consumer will automatically
commit batches _before_ they are processed by the user streams.

An unaware user might accidentally enable auto commit and lose data
during rebalances.

Solves #1289.

## About this PR
📦 Updates
[org.scalameta:scalafmt-core](https://github.com/scalameta/scalafmt)
from `3.8.2` to `3.8.3`

📜 [GitHub Release
Notes](https://github.com/scalameta/scalafmt/releases/tag/v3.8.3) -
[Version
Diff](scalameta/scalafmt@v3.8.2...v3.8.3)

## Usage
✅ **Please merge!**

I'll automatically update this PR to resolve conflicts as long as you
don't change it yourself.

If you'd like to skip this version, you can just close this PR. If you
have any feedback, just mention me in the comments below.

Configure Scala Steward for your repository with a
[`.scala-steward.conf`](https://github.com/scala-steward-org/scala-steward/blob/767fcfecbfd53c507152f6cf15c846176bae561d/docs/repo-specific-configuration.md)
file.

_Have a fantastic day writing Scala!_

<details>
<summary>⚙ Adjust future updates</summary>

Add this to your `.scala-steward.conf` file to ignore future updates of
this dependency:
```
updates.ignore = [ { groupId = "org.scalameta", artifactId = "scalafmt-core" } ]
```
Or, add this to slow down future updates of this dependency:
```
dependencyOverrides = [{
  pullRequests = { frequency = "30 days" },
  dependency = { groupId = "org.scalameta", artifactId = "scalafmt-core" }
}]
```
</details>

<sup>
labels: library-update, early-semver-patch, semver-spec-patch,
commit-count:1
</sup>

Co-authored-by: zio-scala-steward[bot] <145262613+zio-scala-steward[bot]@users.noreply.github.com>

…-control

…-control

…-control

After consumer 1 is shutdown (using stopConsumption), rebalances happen and partitions from consumer 2 are assigned. These streams are never started, so the finalizer completing completedPromise is never called. Waiting for these to complete takes 3 minutes (default maxRebalanceDuration).

In case that streams were assigned and no record was ever put in their queues, there's no need to wait for the stream to complete.

…-control

Co-authored-by: Erik van Oosten <[email protected]>

Co-authored-by: Erik van Oosten <[email protected]>

This caused the "process outstanding commits after a graceful shutdown with aggregateAsync using `maxRebalanceDuration`" to fail.

…-control

…ight from #1360)