Repository: apache/kafka
File: coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java (Java)

Code before:
                        } catch (Throwable ex) {
                            log.error("Failed to load metadata from {} with epoch {} due to {}.",
                                tp, epoch, ex.getMessage(), ex);
                            context.transitionTo(CoordinatorState.FAILED);
                        }
                    } else {
                        log.debug("Failed to complete the loading of metadata for {} in epoch {} since the coordinator does not exist.",
                            tp, epoch);
                    }
                });
            });
        }

        /**
         * Unloads the coordinator.
         */
        private void unload() {
            if (highWatermarklistener != null) {
                partitionWriter.deregisterListener(tp, highWatermarklistener);
                highWatermarklistener = null;
            }
            timer.cancelAll();
            executor.cancelAll();
            deferredEventQueue.failAll(Errors.NOT_COORDINATOR.exception());
            // There is no need to free the current batch, as we will be closing all related resources anyway.
            failCurrentBatch(Errors.NOT_COORDINATOR.exception(), false);
            if (coordinator != null) {
                try {
                    coordinator.onUnloaded();
                } catch (Throwable ex) {
                    log.error("Failed to unload coordinator for {} due to {}.", tp, ex.getMessage(), ex);
                }
            }
            coordinator = null;
        }

        /**
         * Frees the current batch.
         */
        private void freeCurrentBatch() {
            // Cancel the linger timeout.
            currentBatch.lingerTimeoutTask.ifPresent(TimerTask::cancel);

            // Release the buffer only if it is not larger than the maxBatchSize.
            int maxBatchSize = partitionWriter.config(tp).maxMessageSize();

            if (currentBatch.builder.buffer().capacity() <= maxBatchSize) {
                bufferSupplier.release(currentBatch.builder.buffer());
            } else if (currentBatch.buffer.capacity() <= maxBatchSize) {
                bufferSupplier.release(currentBatch.buffer);
            }

Review comment (bug):
Would you mind adding a readable method? for example:
```java
private void failCurrentBatchWithoutRelease(Throwable t) {
        failCurrentBatch(t, false);
    }
```

Code after:
                        } catch (Throwable ex) {
                            log.error("Failed to load metadata from {} with epoch {} due to {}.",
                                tp, epoch, ex.getMessage(), ex);
                            context.transitionTo(CoordinatorState.FAILED);
                        }
                    } else {
                        log.debug("Failed to complete the loading of metadata for {} in epoch {} since the coordinator does not exist.",
                            tp, epoch);
                    }
                });
            });
        }

        /**
         * Unloads the coordinator.
         */
        private void unload() {
            if (highWatermarklistener != null) {
                partitionWriter.deregisterListener(tp, highWatermarklistener);
                highWatermarklistener = null;
            }
            timer.cancelAll();
            executor.cancelAll();
            deferredEventQueue.failAll(Errors.NOT_COORDINATOR.exception());
            failCurrentBatch(Errors.NOT_COORDINATOR.exception());
            if (coordinator != null) {
                try {
                    coordinator.onUnloaded();
                } catch (Throwable ex) {
                    log.error("Failed to unload coordinator for {} due to {}.", tp, ex.getMessage(), ex);
                }
            }
            coordinator = null;
        }

        /**
         * Frees the current batch.
         */
        private void freeCurrentBatch() {
            // Cancel the linger timeout.
            currentBatch.lingerTimeoutTask.ifPresent(TimerTask::cancel);

            // Release the buffer only if it is not larger than the maxBatchSize.
            // We avoid querying the log's configuration for the max message size here,
            // because after a partition leadership change, this throws a NOT_LEADER_OR_FOLLOWER
            // exception. Such exceptions can propagate unexpectedly and disrupt subsequent operations.
            int maxBatchSize = currentBatch.maxBatchSize;

            if (currentBatch.builder.buffer().capacity() <= maxBatchSize) {
                bufferSupplier.release(currentBatch.builder.buffer());
            } else if (currentBatch.buffer.capacity() <= maxBatchSize) {