More heap usage safety back pressure (user controllable? back pressure based on unprocessed messages?)

Nobody has claimed this yet.

We develop Kafka consumer using decaton such as
- Consume messages from Kafka
- Write messages into file and make a large file containing 10K and more messages
- We would like create 100MB+ files, if single message size is 300B and 50% compression ratio, the number of messages contained will be 700K
- Thus we specify `huge decaton.max.pending.records` like 100K, 1M
- We'd like to commit offset for all messages are persisted in the file
- We use external storage and we assume data is persisted when the file descriptor is closed successfully
- So we don't like to commit offset until we close it

As we configure `decaton.max.pending.records` to huge number, decaton could consume the huge number of messages and push down to the processor.
The huge number for decaton.max.pending.records works fine for our application if the consumer consumes messages starting from empty and messages coming lower speed than processor can process.

Assuming the scenario that we have pending records like 100K, 1M for partition and the consumer node fails.
Then partition was rebalanced and decaton will consume messages up to `decaton.max.pending.records` like 100K, 1M faster than the processor can process.
So more than 100K messages might occupy heap usage until the processor processes all messages.
In such a scenario, we actually had OOME for our application.
We might have a similar scenario when restarting the application or resuming the application after stopping for several hours.

So I'm wondering decaton can be improved for this kind of usecase and decaton can have much more better back pressure strategy for heap usage safety.
I propose several possible options
- Processor can control back pressure
- Separate `decaton.max.pending.records` to 2 configurations like the max number of uncommitted messages and the max number of messages yet processed and decaton control back pressure based on these 2 numbers.
- Make decaton consuming speed (rps) configuration and respect it
- Make executor in ProcessUnit configurable and let us control block polling
- https://github.com/line/decaton/blob/d7dd139829d4eb932edaccd019584247ae1d49ee/processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/ProcessorUnit.java#L47
- This executor is used from the thread that calls `poll()`
- If we can block at `executor.execute`, we may control consuming speed

Start with processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/ProcessorUnit.java at the referenced executor and poll() usage. Compare the proposed back-pressure options and define how consumption should be limited while offsets remain uncommitted until persisted files are closed; done should prevent heap exhaustion in the described rebalance and restart scenarios.

Written by the indexing model from the issue text.