From 1eea7f0528954ce8dcbcc4357ae2ef28c1d1e5f2 Mon Sep 17 00:00:00 2001
From: Sean Quah <squah@confluent.io>
Date: Tue, 1 Apr 2025 15:17:39 +0100
Subject: [PATCH] MINOR: Make coordinator runtime inner classes static (#19332)

Make DeferredEventCollection and CoordinatorBatch static classes.
DeferredEventCollection only needs to access the logger and
CoordinatorBatch is only non-static because it holds
DeferredEventCollections.

Reviewers: David Jacot <djacot@confluent.io>
---
 .../common/runtime/CoordinatorRuntime.java    | 20 ++++++++++++++++---
 1 file changed, 17 insertions(+), 3 deletions(-)

diff --git a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java
index 865648da6ab..ebc1868019a 100644
--- a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java
+++ b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java
@@ -459,7 +459,7 @@ public class CoordinatorRuntime<S extends CoordinatorShard<U>, U> implements Aut
      * A simple container class to hold all the attributes
      * related to a pending batch.
      */
-    private class CoordinatorBatch {
+    private static class CoordinatorBatch {
         /**
          * The base (or first) offset of the batch. If the batch fails
          * for any reason, the state machines is rolled back to it.
@@ -510,6 +510,7 @@ public class CoordinatorRuntime<S extends CoordinatorShard<U>, U> implements Aut
         long nextOffset;
 
         CoordinatorBatch(
+            Logger log,
             long baseOffset,
             long appendTimeMs,
             int maxBatchSize,
@@ -526,7 +527,7 @@ public class CoordinatorRuntime<S extends CoordinatorShard<U>, U> implements Aut
             this.buffer = buffer;
             this.builder = builder;
             this.lingerTimeoutTask = lingerTimeoutTask;
-            this.deferredEvents = new DeferredEventCollection();
+            this.deferredEvents = new DeferredEventCollection(log);
         }
     }
 
@@ -893,6 +894,7 @@ public class CoordinatorRuntime<S extends CoordinatorShard<U>, U> implements Aut
                 }
 
                 currentBatch = new CoordinatorBatch(
+                    log,
                     prevLastWrittenOffset,
                     currentTimeMs,
                     maxBatchSize,
@@ -1160,9 +1162,21 @@ public class CoordinatorRuntime<S extends CoordinatorShard<U>, U> implements Aut
      * A collection of {@link DeferredEvent}. When completed, completes all the events in the collection
      * and logs any exceptions thrown.
      */
-    class DeferredEventCollection implements DeferredEvent {
+    static class DeferredEventCollection implements DeferredEvent {
+        /**
+         * The logger.
+         */
+        private final Logger log;
+
+        /**
+         * The list of events.
+         */
         private final List<DeferredEvent> events = new ArrayList<>();
 
+        public DeferredEventCollection(Logger log) {
+            this.log = log;
+        }
+
         @Override
         public void complete(Throwable t) {
             for (DeferredEvent event : events) {