fix(inkless:consume): stop a slow ListOffsets from hanging the broker [KC-537] - #798

… [KC-537] (#798)

* fix(inkless:control-plane): don't serialize listOffsets across callers

AbstractControlPlane.listOffsets was synchronized, so every ListOffsets
on the broker waited for the slowest single control-plane query. The
lock only existed for InMemoryControlPlane, whose listOffset was the one
implementation without its own synchronization. Move the lock there,
matching findBatchesForExistingPartition, so PostgresControlPlane runs
queries concurrently.

* fix(inkless:consume): bound the fetch offset and delete records pools

FetchOffsetHandler and DeleteRecordsInterceptor used cached thread
pools, which create one thread per submission when no worker is idle.
A slow control plane turned every arriving request into a parked OS
thread with no upper bound.

Replace both with fixed pools and a bounded queue. A submission that
finds the queue full fails on the caller with RejectedExecutionException,
which the handler turns into a retriable REQUEST_TIMED_OUT response
instead of letting the exception escape to the request handler. The
fetch offset pool size is set by the new fetch.offset.thread.pool.size
config (default 8), with a queue of 100 slots per thread. Rejections are
counted by the new FetchOffsetRejectedRate metric. DeleteRecords is an
admin path with a low arrival rate and uses a fixed pool of 2 threads
with a queue of 200 slots. Both pools report queue depth and active
threads through ThreadPoolMonitor.

A task that purgatory cancels while it is still queued keeps its queue
slot until a worker reaches it and skips it. This is the same behavior
as the remote log reader pool in Kafka. During a control-plane stall,
cancelled requests can therefore still fill the queue, and new
submissions are rejected until the workers catch up.

Assisted-by: Claude Opus 5.5

… [KC-537] (#798)

* fix(inkless:control-plane): don't serialize listOffsets across callers

AbstractControlPlane.listOffsets was synchronized, so every ListOffsets
on the broker waited for the slowest single control-plane query. The
lock only existed for InMemoryControlPlane, whose listOffset was the one
implementation without its own synchronization. Move the lock there,
matching findBatchesForExistingPartition, so PostgresControlPlane runs
queries concurrently.

* fix(inkless:consume): bound the fetch offset and delete records pools

FetchOffsetHandler and DeleteRecordsInterceptor used cached thread
pools, which create one thread per submission when no worker is idle.
A slow control plane turned every arriving request into a parked OS
thread with no upper bound.

Replace both with fixed pools and a bounded queue. A submission that
finds the queue full fails on the caller with RejectedExecutionException,
which the handler turns into a retriable REQUEST_TIMED_OUT response
instead of letting the exception escape to the request handler. The
fetch offset pool size is set by the new fetch.offset.thread.pool.size
config (default 8), with a queue of 100 slots per thread. Rejections are
counted by the new FetchOffsetRejectedRate metric. DeleteRecords is an
admin path with a low arrival rate and uses a fixed pool of 2 threads
with a queue of 200 slots. Both pools report queue depth and active
threads through ThreadPoolMonitor.

A task that purgatory cancels while it is still queued keeps its queue
slot until a worker reaches it and skips it. This is the same behavior
as the remote log reader pool in Kafka. During a control-plane stall,
cancelled requests can therefore still fill the queue, and new
submissions are rejected until the workers catch up.

Assisted-by: Claude Opus 5.5

… [KC-537] (#798)

* fix(inkless:control-plane): don't serialize listOffsets across callers

AbstractControlPlane.listOffsets was synchronized, so every ListOffsets
on the broker waited for the slowest single control-plane query. The
lock only existed for InMemoryControlPlane, whose listOffset was the one
implementation without its own synchronization. Move the lock there,
matching findBatchesForExistingPartition, so PostgresControlPlane runs
queries concurrently.

* fix(inkless:consume): bound the fetch offset and delete records pools

FetchOffsetHandler and DeleteRecordsInterceptor used cached thread
pools, which create one thread per submission when no worker is idle.
A slow control plane turned every arriving request into a parked OS
thread with no upper bound.

Replace both with fixed pools and a bounded queue. A submission that
finds the queue full fails on the caller with RejectedExecutionException,
which the handler turns into a retriable REQUEST_TIMED_OUT response
instead of letting the exception escape to the request handler. The
fetch offset pool size is set by the new fetch.offset.thread.pool.size
config (default 8), with a queue of 100 slots per thread. Rejections are
counted by the new FetchOffsetRejectedRate metric. DeleteRecords is an
admin path with a low arrival rate and uses a fixed pool of 2 threads
with a queue of 200 slots. Both pools report queue depth and active
threads through ThreadPoolMonitor.

A task that purgatory cancels while it is still queued keeps its queue
slot until a worker reaches it and skips it. This is the same behavior
as the remote log reader pool in Kafka. During a control-plane stall,
cancelled requests can therefore still fill the queue, and new
submissions are rejected until the workers catch up.

Assisted-by: Claude Opus 5.5