This is an automated email from the ASF dual-hosted git repository.
squah-confluent pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 03910980ca5 MINOR: Code cleanups in coordinator-common runtime (#22660)
03910980ca5 is described below
commit 03910980ca58ff5a3f34bb5c6a49a22fca53a22e
Author: Sanskar Jhajharia <[email protected]>
AuthorDate: Fri Jun 26 17:24:35 2026 +0530
MINOR: Code cleanups in coordinator-common runtime (#22660)
Minor, behavior-preserving cleanups across the `coordinator-common`
runtime package. No functional changes. The changes fall into two
categories:
1. **Replace `Arrays.asList(...)` with `List.of(...)`** in production
and test code where the list is only read. This is more idiomatic,
returns a truly immutable list, and lets us drop the `java.util.Arrays`
import in favor of `java.util.List`.
2. **Fix a stray quote** in a comment in `CoordinatorLoaderImpl`
Reviewers: Sean Quah <[email protected]>
---
.../common/runtime/CoordinatorLoaderImpl.java | 2 +-
.../common/runtime/CoordinatorRuntime.java | 6 ++--
.../runtime/CoordinatorRuntimeMetricsImpl.java | 4 +--
.../common/runtime/CoordinatorLoaderImplTest.java | 41 +++++++++++-----------
.../common/runtime/EventAccumulatorTest.java | 2 +-
.../runtime/MultiThreadedEventProcessorTest.java | 9 +++--
.../common/runtime/MetadataImageBuilder.java | 4 +--
7 files changed, 33 insertions(+), 35 deletions(-)
diff --git
a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImpl.java
b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImpl.java
index 2616e8f0ccf..a64a3076173 100644
---
a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImpl.java
+++
b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImpl.java
@@ -197,7 +197,7 @@ public class CoordinatorLoaderImpl<T> implements
CoordinatorLoader<T> {
int sizeInBytes = fileRecords.sizeInBytes();
int bytesNeeded = Math.max(loadBufferSize, sizeInBytes);
- // "minOneMessage = true in the above log.read() means that the
buffer may need to
+ // minOneMessage = true in the above log.read() means that the
buffer may need to
// be grown to ensure progress can be made.
if (buffer.capacity() < bytesNeeded) {
if (loadBufferSize < bytesNeeded) {
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 f317ebe31b9..16b9fd4a906 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
@@ -583,7 +583,7 @@ public class CoordinatorRuntime<S extends
CoordinatorShard<U>, U> implements Aut
loader.load(tp, coordinator).whenComplete((summary, exception) -> {
scheduleInternalOperation("CompleteLoad(tp=" + tp + ", epoch="
+ epoch + ")", tp, () -> {
CoordinatorContext context = coordinators.get(tp);
- if (context != null) {
+ if (context != null) {
if (context.state != CoordinatorState.LOADING) {
log.info("Ignored load completion from {} because
context is in {} state.",
context.tp, context.state);
@@ -872,9 +872,9 @@ public class CoordinatorRuntime<S extends
CoordinatorShard<U>, U> implements Aut
}
/**
- * Completes the given event once all pending writes are completed.
+ * Completes the given event once all pending writes are completed.
*
- * @param event The event to complete once all pending
+ * @param event The event to complete once all pending
* writes are completed.
*/
private void waitForPendingWrites(DeferredEvent event) {
diff --git
a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImpl.java
b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImpl.java
index 8835455cb28..b681e9d2bcf 100644
---
a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImpl.java
+++
b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImpl.java
@@ -27,7 +27,7 @@ import org.apache.kafka.common.metrics.stats.Rate;
import org.apache.kafka.common.metrics.stats.WindowedCount;
import
org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime.CoordinatorState;
-import java.util.Arrays;
+import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@@ -346,7 +346,7 @@ public class CoordinatorRuntimeMetricsImpl implements
CoordinatorRuntimeMetrics
@Override
public void close() {
- Arrays.asList(
+ List.of(
numPartitionsLoading,
numPartitionsActive,
numPartitionsFailed,
diff --git
a/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImplTest.java
b/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImplTest.java
index 54b996025ad..cb6c2aa837c 100644
---
a/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImplTest.java
+++
b/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImplTest.java
@@ -41,7 +41,6 @@ import org.mockito.invocation.InvocationOnMock;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -139,7 +138,7 @@ class CoordinatorLoaderImplTest {
when(log.logStartOffset()).thenReturn(0L);
when(log.highWatermark()).thenReturn(0L);
- FetchDataInfo readResult1 = logReadResult(0, Arrays.asList(
+ FetchDataInfo readResult1 = logReadResult(0, List.of(
new SimpleRecord("k1".getBytes(), "v1".getBytes()),
new SimpleRecord("k2".getBytes(), "v2".getBytes())
));
@@ -147,7 +146,7 @@ class CoordinatorLoaderImplTest {
when(log.read(0L, 1000, FetchIsolation.LOG_END, true))
.thenReturn(readResult1);
- FetchDataInfo readResult2 = logReadResult(2, Arrays.asList(
+ FetchDataInfo readResult2 = logReadResult(2, List.of(
new SimpleRecord("k3".getBytes(), "v3".getBytes()),
new SimpleRecord("k4".getBytes(), "v4".getBytes()),
new SimpleRecord("k5".getBytes(), "v5".getBytes())
@@ -156,7 +155,7 @@ class CoordinatorLoaderImplTest {
when(log.read(2L, 1000, FetchIsolation.LOG_END, true))
.thenReturn(readResult2);
- FetchDataInfo readResult3 = logReadResult(5, 100L, (short) 5,
Arrays.asList(
+ FetchDataInfo readResult3 = logReadResult(5, 100L, (short) 5,
List.of(
new SimpleRecord("k6".getBytes(), "v6".getBytes()),
new SimpleRecord("k7".getBytes(), "v7".getBytes())
));
@@ -225,7 +224,7 @@ class CoordinatorLoaderImplTest {
)) {
when(log.logStartOffset()).thenReturn(0L);
- FetchDataInfo readResult = logReadResult(0, Arrays.asList(
+ FetchDataInfo readResult = logReadResult(0, List.of(
new SimpleRecord("k1".getBytes(), "v1".getBytes()),
new SimpleRecord("k2".getBytes(), "v2".getBytes())
));
@@ -271,7 +270,7 @@ class CoordinatorLoaderImplTest {
)) {
when(log.logStartOffset()).thenReturn(0L);
- FetchDataInfo readResult = logReadResult(0, Arrays.asList(
+ FetchDataInfo readResult = logReadResult(0, List.of(
new SimpleRecord("k1".getBytes(), "v1".getBytes()),
new SimpleRecord("k2".getBytes(), "v2".getBytes())
));
@@ -308,7 +307,7 @@ class CoordinatorLoaderImplTest {
)) {
when(log.logStartOffset()).thenReturn(0L);
- FetchDataInfo readResult = logReadResult(0, Arrays.asList(
+ FetchDataInfo readResult = logReadResult(0, List.of(
new SimpleRecord("k1".getBytes(), "v1".getBytes()),
new SimpleRecord("k2".getBytes(), "v2".getBytes())
));
@@ -378,7 +377,7 @@ class CoordinatorLoaderImplTest {
long startTimeMs = time.milliseconds();
when(log.logStartOffset()).thenReturn(0L);
- FetchDataInfo readResult1 = logReadResult(0, Arrays.asList(
+ FetchDataInfo readResult1 = logReadResult(0, List.of(
new SimpleRecord("k1".getBytes(), "v1".getBytes()),
new SimpleRecord("k2".getBytes(), "v2".getBytes())
));
@@ -389,7 +388,7 @@ class CoordinatorLoaderImplTest {
return readResult1;
});
- FetchDataInfo readResult2 = logReadResult(2, Arrays.asList(
+ FetchDataInfo readResult2 = logReadResult(2, List.of(
new SimpleRecord("k3".getBytes(), "v3".getBytes()),
new SimpleRecord("k4".getBytes(), "v4".getBytes()),
new SimpleRecord("k5".getBytes(), "v5".getBytes())
@@ -426,7 +425,7 @@ class CoordinatorLoaderImplTest {
when(log.logStartOffset()).thenReturn(0L);
when(log.highWatermark()).thenReturn(0L, 0L, 2L);
- FetchDataInfo readResult1 = logReadResult(0, Arrays.asList(
+ FetchDataInfo readResult1 = logReadResult(0, List.of(
new SimpleRecord("k1".getBytes(), "v1".getBytes()),
new SimpleRecord("k2".getBytes(), "v2".getBytes())
));
@@ -434,7 +433,7 @@ class CoordinatorLoaderImplTest {
when(log.read(0L, 1000, FetchIsolation.LOG_END, true))
.thenReturn(readResult1);
- FetchDataInfo readResult2 = logReadResult(2, Arrays.asList(
+ FetchDataInfo readResult2 = logReadResult(2, List.of(
new SimpleRecord("k3".getBytes(), "v3".getBytes()),
new SimpleRecord("k4".getBytes(), "v4".getBytes()),
new SimpleRecord("k5".getBytes(), "v5".getBytes())
@@ -443,7 +442,7 @@ class CoordinatorLoaderImplTest {
when(log.read(2L, 1000, FetchIsolation.LOG_END, true))
.thenReturn(readResult2);
- FetchDataInfo readResult3 = logReadResult(5, Arrays.asList(
+ FetchDataInfo readResult3 = logReadResult(5, List.of(
new SimpleRecord("k6".getBytes(), "v6".getBytes()),
new SimpleRecord("k7".getBytes(), "v7".getBytes())
));
@@ -517,7 +516,7 @@ class CoordinatorLoaderImplTest {
when(log.logStartOffset()).thenReturn(0L);
when(log.highWatermark()).thenReturn(5L, 7L, 7L);
- FetchDataInfo readResult1 = logReadResult(0, Arrays.asList(
+ FetchDataInfo readResult1 = logReadResult(0, List.of(
new SimpleRecord("k1".getBytes(), "v1".getBytes()),
new SimpleRecord("k2".getBytes(), "v2".getBytes())
));
@@ -525,7 +524,7 @@ class CoordinatorLoaderImplTest {
when(log.read(0L, 1000, FetchIsolation.LOG_END, true))
.thenReturn(readResult1);
- FetchDataInfo readResult2 = logReadResult(2, Arrays.asList(
+ FetchDataInfo readResult2 = logReadResult(2, List.of(
new SimpleRecord("k3".getBytes(), "v3".getBytes()),
new SimpleRecord("k4".getBytes(), "v4".getBytes()),
new SimpleRecord("k5".getBytes(), "v5".getBytes())
@@ -534,7 +533,7 @@ class CoordinatorLoaderImplTest {
when(log.read(2L, 1000, FetchIsolation.LOG_END, true))
.thenReturn(readResult2);
- FetchDataInfo readResult3 = logReadResult(5, Arrays.asList(
+ FetchDataInfo readResult3 = logReadResult(5, List.of(
new SimpleRecord("k6".getBytes(), "v6".getBytes()),
new SimpleRecord("k7".getBytes(), "v7".getBytes())
));
@@ -582,7 +581,7 @@ class CoordinatorLoaderImplTest {
when(log.logStartOffset()).thenReturn(0L);
when(log.highWatermark()).thenReturn(7L);
- FetchDataInfo readResult1 = logReadResult(0, Arrays.asList(
+ FetchDataInfo readResult1 = logReadResult(0, List.of(
new SimpleRecord("k1".getBytes(), "v1".getBytes()),
new SimpleRecord("k2".getBytes(), "v2".getBytes())
));
@@ -590,7 +589,7 @@ class CoordinatorLoaderImplTest {
when(log.read(0L, 1000, FetchIsolation.LOG_END, true))
.thenReturn(readResult1);
- FetchDataInfo readResult2 = logReadResult(2, Arrays.asList(
+ FetchDataInfo readResult2 = logReadResult(2, List.of(
new SimpleRecord("k3".getBytes(), "v3".getBytes()),
new SimpleRecord("k4".getBytes(), "v4".getBytes()),
new SimpleRecord("k5".getBytes(), "v5".getBytes())
@@ -599,14 +598,14 @@ class CoordinatorLoaderImplTest {
when(log.read(2L, 1000, FetchIsolation.LOG_END, true))
.thenReturn(readResult2);
- FetchDataInfo readResult3 = logReadResult(5, Arrays.asList(
+ FetchDataInfo readResult3 = logReadResult(5, List.of(
new SimpleRecord("k6".getBytes(), "v6".getBytes())
));
when(log.read(5L, 1000, FetchIsolation.LOG_END, true))
.thenReturn(readResult3);
- FetchDataInfo readResult4 = logReadResult(6, Arrays.asList(
+ FetchDataInfo readResult4 = logReadResult(6, List.of(
new SimpleRecord("k7".getBytes(), "v7".getBytes())
));
@@ -656,7 +655,7 @@ class CoordinatorLoaderImplTest {
when(log.highWatermark()).thenReturn(0L);
when(partitionLogEndOffsetSupplier.apply(tp)).thenReturn(Optional.of(5L)).thenReturn(Optional.of(-1L));
- FetchDataInfo readResult1 = logReadResult(0, Arrays.asList(
+ FetchDataInfo readResult1 = logReadResult(0, List.of(
new SimpleRecord("k1".getBytes(), "v1".getBytes()),
new SimpleRecord("k2".getBytes(), "v2".getBytes())
));
@@ -664,7 +663,7 @@ class CoordinatorLoaderImplTest {
when(log.read(0L, 1000, FetchIsolation.LOG_END, true))
.thenReturn(readResult1);
- FetchDataInfo readResult2 = logReadResult(2, Arrays.asList(
+ FetchDataInfo readResult2 = logReadResult(2, List.of(
new SimpleRecord("k3".getBytes(), "v3".getBytes()),
new SimpleRecord("k4".getBytes(), "v4".getBytes()),
new SimpleRecord("k5".getBytes(), "v5".getBytes())
diff --git
a/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/EventAccumulatorTest.java
b/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/EventAccumulatorTest.java
index d35a0776db0..6bbf0e44e13 100644
---
a/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/EventAccumulatorTest.java
+++
b/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/EventAccumulatorTest.java
@@ -82,7 +82,7 @@ public class EventAccumulatorTest {
assertEquals(0, accumulator.size());
assertNull(accumulator.poll());
- List<MockEvent> events = Arrays.asList(
+ List<MockEvent> events = List.of(
new MockEvent(1, 0),
new MockEvent(1, 1),
new MockEvent(1, 2),
diff --git
a/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/MultiThreadedEventProcessorTest.java
b/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/MultiThreadedEventProcessorTest.java
index 751dc0c820b..35281afe1a7 100644
---
a/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/MultiThreadedEventProcessorTest.java
+++
b/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/MultiThreadedEventProcessorTest.java
@@ -26,7 +26,6 @@ import org.junit.jupiter.api.Timeout;
import org.mockito.ArgumentCaptor;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
@@ -178,7 +177,7 @@ public class MultiThreadedEventProcessorTest {
)) {
AtomicInteger numEventsExecuted = new AtomicInteger(0);
- List<FutureEvent<Integer>> events = Arrays.asList(
+ List<FutureEvent<Integer>> events = List.of(
new FutureEvent<>(new TopicPartition("foo", 0),
numEventsExecuted::incrementAndGet),
new FutureEvent<>(new TopicPartition("foo", 1),
numEventsExecuted::incrementAndGet),
new FutureEvent<>(new TopicPartition("foo", 2),
numEventsExecuted::incrementAndGet),
@@ -215,7 +214,7 @@ public class MultiThreadedEventProcessorTest {
)) {
AtomicInteger numEventsExecuted = new AtomicInteger(0);
- List<FutureEvent<Integer>> events = Arrays.asList(
+ List<FutureEvent<Integer>> events = List.of(
new FutureEvent<>(new TopicPartition("foo", 0),
numEventsExecuted::incrementAndGet, true), // Event 0
new FutureEvent<>(new TopicPartition("foo", 1),
numEventsExecuted::incrementAndGet, true), // Event 1
new FutureEvent<>(new TopicPartition("foo", 0),
numEventsExecuted::incrementAndGet, true), // Event 2
@@ -323,7 +322,7 @@ public class MultiThreadedEventProcessorTest {
true
);
- List<FutureEvent<Integer>> events = Arrays.asList(
+ List<FutureEvent<Integer>> events = List.of(
new FutureEvent<>(new TopicPartition("foo", 0),
numEventsExecuted::incrementAndGet),
new FutureEvent<>(new TopicPartition("foo", 0),
numEventsExecuted::incrementAndGet),
new FutureEvent<>(new TopicPartition("foo", 0),
numEventsExecuted::incrementAndGet),
@@ -482,7 +481,7 @@ public class MultiThreadedEventProcessorTest {
return null;
}).when(mockRuntimeMetrics).recordThreadIdleTime(idleTimeCaptured.capture());
- List<FutureEvent<Integer>> events = Arrays.asList(
+ List<FutureEvent<Integer>> events = List.of(
new FutureEvent<>(new TopicPartition("foo", 0),
numEventsExecuted::incrementAndGet),
new FutureEvent<>(new TopicPartition("foo", 1),
numEventsExecuted::incrementAndGet),
new FutureEvent<>(new TopicPartition("foo", 2),
numEventsExecuted::incrementAndGet),
diff --git
a/coordinator-common/src/testFixtures/java/org/apache/kafka/coordinator/common/runtime/MetadataImageBuilder.java
b/coordinator-common/src/testFixtures/java/org/apache/kafka/coordinator/common/runtime/MetadataImageBuilder.java
index e2eacf9f537..3e417f8f196 100644
---
a/coordinator-common/src/testFixtures/java/org/apache/kafka/coordinator/common/runtime/MetadataImageBuilder.java
+++
b/coordinator-common/src/testFixtures/java/org/apache/kafka/coordinator/common/runtime/MetadataImageBuilder.java
@@ -24,7 +24,7 @@ import org.apache.kafka.image.MetadataDelta;
import org.apache.kafka.image.MetadataImage;
import org.apache.kafka.image.MetadataProvenance;
-import java.util.Arrays;
+import java.util.List;
public class MetadataImageBuilder {
private final MetadataDelta delta;
@@ -52,7 +52,7 @@ public class MetadataImageBuilder {
delta.replay(new PartitionRecord()
.setTopicId(topicId)
.setPartitionId(i)
- .setReplicas(Arrays.asList(i % 4, (i + 1) % 4)));
+ .setReplicas(List.of(i % 4, (i + 1) % 4)));
}
return this;
}
