From 59efa12d0c0c098e6f57848bf5a43b459c0563e0 Mon Sep 17 00:00:00 2001
From: belugabehr <12578579+belugabehr@users.noreply.github.com>
Date: Wed, 20 May 2020 23:05:28 -0400
Subject: [PATCH] KAFKA-9409: Supplement immutability of ClusterConfigState
 class in Connect (#7942)

The class claims to be immutable, but there are some mutable features of this class.
Increase the immutability of it and add a little cleanup:

* Pre-initialize size of ArrayList
* Remove superfluous syntax
* Use ArrayList instead of LinkedList since the list is created once

Reviewers: Ron Dagostino <rdagostino@confluent.io>, Konstantine Karantasis <konstantine@confluent.io>
---
 .../distributed/ClusterConfigState.java       | 23 ++++++++++---------
 1 file changed, 12 insertions(+), 11 deletions(-)

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/ClusterConfigState.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/ClusterConfigState.java
index a85a8e69e9f..717120d8508 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/ClusterConfigState.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/ClusterConfigState.java
@@ -24,7 +24,6 @@ import org.apache.kafka.connect.util.ConnectorTaskId;
 
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -39,11 +38,11 @@ public class ClusterConfigState {
     public static final ClusterConfigState EMPTY = new ClusterConfigState(
             NO_OFFSET,
             null,
-            Collections.<String, Integer>emptyMap(),
-            Collections.<String, Map<String, String>>emptyMap(),
-            Collections.<String, TargetState>emptyMap(),
-            Collections.<ConnectorTaskId, Map<String, String>>emptyMap(),
-            Collections.<String>emptySet());
+            Collections.emptyMap(),
+            Collections.emptyMap(),
+            Collections.emptyMap(),
+            Collections.emptyMap(),
+            Collections.emptySet());
 
     private final long offset;
     private final SessionKey sessionKey;
@@ -190,7 +189,7 @@ public class ClusterConfigState {
                 taskConfigs.put(taskConfigEntry.getKey().task(), configs);
             }
         }
-        return new LinkedList<>(taskConfigs.values());
+        return Collections.unmodifiableList(new ArrayList<>(taskConfigs.values()));
     }
 
     /**
@@ -209,19 +208,21 @@ public class ClusterConfigState {
      * @return the current set of connector task IDs
      */
     public List<ConnectorTaskId> tasks(String connectorName) {
-        if (inconsistentConnectors.contains(connectorName))
+        if (inconsistentConnectors.contains(connectorName)) {
             return Collections.emptyList();
+        }
 
         Integer numTasks = connectorTaskCounts.get(connectorName);
-        if (numTasks == null)
+        if (numTasks == null) {
             return Collections.emptyList();
+        }
 
-        List<ConnectorTaskId> taskIds = new ArrayList<>();
+        List<ConnectorTaskId> taskIds = new ArrayList<>(numTasks);
         for (int taskIndex = 0; taskIndex < numTasks; taskIndex++) {
             ConnectorTaskId taskId = new ConnectorTaskId(connectorName, taskIndex);
             taskIds.add(taskId);
         }
-        return taskIds;
+        return Collections.unmodifiableList(taskIds);
     }
 
     /**