KAFKA-19519: Introduce group.coordinator.append.max.buffer.size config

[DL1231]

Changes

1. New Dynamic Configurations

• group.coordinator.append.max.buffer.size: Largest buffer size allowed by GroupCoordinator
• share.coordinator.append.max.buffer.size: Largest buffer size allowed by ShareCoordinator

Both configurations default to 1 * 1024 * 1024 + Records.LOG_OVERHEAD with minimum value of 512 * 1024.

2. Extended CoordinatorRuntime Builder Interface

Added withMaxBufferSize(Supplier maxBufferSizeSupplier) method to allow different coordinator implementations to supply their buffer size configuration.

3. New Monitoring Metrics

• coordinator-append-buffer-size-bytes: Current total size in bytes of the append buffers being held in the coordinator's cache
• coordinator-append-buffer-skip-cache-count: Count of oversized append buffers that were discarded instead of being cached upon release

[squah-confluent]

Thanks for the patch! I left a few comments. I haven't reviewed the full PR yet.

[squah-confluent]

The size implementations aren't thread safe, but they will be called from a metrics thread.

[DL1231]

Thank you for pointing out this issue. As BufferSupplier is designed to be a "simple non-threadsafe interface for caching byte buffers," I've made the following targeted changes to address the specific concerns with the size() method when called from metrics threads:

• Changed the bufferMap in DefaultSupplier to a ConcurrentHashMap to prevent potential ConcurrentModificationException.
• Added the volatile keyword to cachedBuffer in GrowableBufferSupplier to resolve visibility issues.

I intentionally avoided adding synchronization to the get(), release(), and size() methods to maintain performance, as comprehensive thread safety isn't a design goal for this interface. These changes specifically address the metrics collection use case while preserving the lightweight. WDYT?

[squah-confluent]

Thanks for updating the PR!

Avoiding locks is the right thing to do but the fixes do not look complete.

• The ArrayDeques within DefaultSupplier suffer from the same visibility issue as cachedBuffer.
• The first read of cachedBuffer within GrowableBufferSupplier.size can be non-null while the second read can be null.

Instead I would propose we introduce a size AtomicLong and update them in get and release.