Convert AllNodes to a Java record

dain · dain · commit a844f0db5041 · 2025-07-01T20:00:48.000-07:00
diff --git a/core/trino-main/src/main/java/io/trino/connector/CatalogPruneTask.java b/core/trino-main/src/main/java/io/trino/connector/CatalogPruneTask.java
@@ -128,7 +128,7 @@ public ThreadPoolExecutorMBean getExecutor()
     public void pruneWorkerCatalogs()
     {
         AllNodes allNodes = internalNodeManager.getAllNodes();
-        Set<URI> online = Stream.of(allNodes.getActiveNodes(), allNodes.getInactiveNodes(), allNodes.getDrainingNodes(), allNodes.getDrainedNodes(), allNodes.getShuttingDownNodes())
+        Set<URI> online = Stream.of(allNodes.activeNodes(), allNodes.inactiveNodes(), allNodes.drainingNodes(), allNodes.drainedNodes(), allNodes.shuttingDownNodes())
                 .flatMap(Set::stream)
                 .map(InternalNode::getInternalUri)
                 .filter(uri -> !uri.equals(currentNode.getInternalUri()))
diff --git a/core/trino-main/src/main/java/io/trino/connector/system/NodeSystemTable.java b/core/trino-main/src/main/java/io/trino/connector/system/NodeSystemTable.java
@@ -80,11 +80,11 @@ public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, Connect
     {
         Builder table = InMemoryRecordSet.builder(NODES_TABLE);
         AllNodes allNodes = nodeManager.getAllNodes();
-        addRows(table, allNodes.getActiveNodes(), ACTIVE);
-        addRows(table, allNodes.getInactiveNodes(), INACTIVE);
-        addRows(table, allNodes.getShuttingDownNodes(), SHUTTING_DOWN);
-        addRows(table, allNodes.getDrainingNodes(), DRAINING);
-        addRows(table, allNodes.getDrainedNodes(), DRAINED);
+        addRows(table, allNodes.activeNodes(), ACTIVE);
+        addRows(table, allNodes.inactiveNodes(), INACTIVE);
+        addRows(table, allNodes.shuttingDownNodes(), SHUTTING_DOWN);
+        addRows(table, allNodes.drainingNodes(), DRAINING);
+        addRows(table, allNodes.drainedNodes(), DRAINED);
 
         return table.build().cursor();
     }
diff --git a/core/trino-main/src/main/java/io/trino/cost/TaskCountEstimator.java b/core/trino-main/src/main/java/io/trino/cost/TaskCountEstimator.java
@@ -42,7 +42,7 @@ public TaskCountEstimator(NodeSchedulerConfig nodeSchedulerConfig, InternalNodeM
         boolean schedulerIncludeCoordinator = nodeSchedulerConfig.isIncludeCoordinator();
         requireNonNull(nodeManager, "nodeManager is null");
         this.numberOfNodes = () -> {
-            Set<InternalNode> activeNodes = nodeManager.getAllNodes().getActiveNodes();
+            Set<InternalNode> activeNodes = nodeManager.getAllNodes().activeNodes();
             int count;
             if (schedulerIncludeCoordinator) {
                 count = activeNodes.size();
diff --git a/core/trino-main/src/main/java/io/trino/execution/ClusterSizeMonitor.java b/core/trino-main/src/main/java/io/trino/execution/ClusterSizeMonitor.java
@@ -137,10 +137,10 @@ private synchronized void removeFuture(MinNodesFuture minNodesFuture)
     private synchronized void updateAllNodes(AllNodes allNodes)
     {
         if (includeCoordinator) {
-            currentCount = allNodes.getActiveNodes().size();
+            currentCount = allNodes.activeNodes().size();
         }
         else {
-            currentCount = Sets.difference(allNodes.getActiveNodes(), allNodes.getActiveCoordinators()).size();
+            currentCount = Sets.difference(allNodes.activeNodes(), allNodes.activeCoordinators()).size();
         }
 
         ImmutableList.Builder<SettableFuture<Void>> listenersBuilder = ImmutableList.builder();
diff --git a/core/trino-main/src/main/java/io/trino/execution/scheduler/faulttolerant/EventDrivenTaskSourceFactory.java b/core/trino-main/src/main/java/io/trino/execution/scheduler/faulttolerant/EventDrivenTaskSourceFactory.java
@@ -239,7 +239,7 @@ private SplitAssigner createSplitAssigner(
                     outputDataSizeEstimates,
                     fragment,
                     getFaultTolerantExecutionHashDistributionComputeTaskTargetSize(session).toBytes(),
-                    toIntExact(round(getFaultTolerantExecutionHashDistributionComputeTasksToNodesMinRatio(session) * nodeManager.getAllNodes().getActiveNodes().size())),
+                    toIntExact(round(getFaultTolerantExecutionHashDistributionComputeTasksToNodesMinRatio(session) * nodeManager.getAllNodes().activeNodes().size())),
                     Integer.MAX_VALUE); // compute tasks are bounded by the number of partitions anyways
         }
         if (partitioning.equals(SCALED_WRITER_HASH_DISTRIBUTION)
@@ -253,7 +253,7 @@ private SplitAssigner createSplitAssigner(
                     outputDataSizeEstimates,
                     fragment,
                     getFaultTolerantExecutionHashDistributionWriteTaskTargetSize(session).toBytes(),
-                    toIntExact(round(getFaultTolerantExecutionHashDistributionWriteTasksToNodesMinRatio(session) * nodeManager.getAllNodes().getActiveNodes().size())),
+                    toIntExact(round(getFaultTolerantExecutionHashDistributionWriteTasksToNodesMinRatio(session) * nodeManager.getAllNodes().activeNodes().size())),
                     getFaultTolerantExecutionHashDistributionWriteTaskTargetMaxCount(session));
         }
 
diff --git a/core/trino-main/src/main/java/io/trino/node/AllNodes.java b/core/trino-main/src/main/java/io/trino/node/AllNodes.java
@@ -15,86 +15,25 @@
 
 import com.google.common.collect.ImmutableSet;
 
-import java.util.Objects;
 import java.util.Set;
 
 import static java.util.Objects.requireNonNull;
 
-public class AllNodes
+public record AllNodes(
+        Set<InternalNode> activeNodes,
+        Set<InternalNode> inactiveNodes,
+        Set<InternalNode> drainingNodes,
+        Set<InternalNode> drainedNodes,
+        Set<InternalNode> shuttingDownNodes,
+        Set<InternalNode> activeCoordinators)
 {
-    private final Set<InternalNode> activeNodes;
-    private final Set<InternalNode> inactiveNodes;
-    private final Set<InternalNode> drainingNodes;
-    private final Set<InternalNode> drainedNodes;
-    private final Set<InternalNode> shuttingDownNodes;
-    private final Set<InternalNode> activeCoordinators;
-
-    public AllNodes(Set<InternalNode> activeNodes,
-            Set<InternalNode> inactiveNodes,
-            Set<InternalNode> drainingNodes,
-            Set<InternalNode> drainedNodes,
-            Set<InternalNode> shuttingDownNodes,
-            Set<InternalNode> activeCoordinators)
-    {
-        this.activeNodes = ImmutableSet.copyOf(requireNonNull(activeNodes, "activeNodes is null"));
-        this.inactiveNodes = ImmutableSet.copyOf(requireNonNull(inactiveNodes, "inactiveNodes is null"));
-        this.drainedNodes = ImmutableSet.copyOf(requireNonNull(drainedNodes, "drainedNodes is null"));
-        this.drainingNodes = ImmutableSet.copyOf(requireNonNull(drainingNodes, "drainingNodes is null"));
-        this.shuttingDownNodes = ImmutableSet.copyOf(requireNonNull(shuttingDownNodes, "shuttingDownNodes is null"));
-        this.activeCoordinators = ImmutableSet.copyOf(requireNonNull(activeCoordinators, "activeCoordinators is null"));
-    }
-
-    public Set<InternalNode> getActiveNodes()
-    {
-        return activeNodes;
-    }
-
-    public Set<InternalNode> getInactiveNodes()
-    {
-        return inactiveNodes;
-    }
-
-    public Set<InternalNode> getShuttingDownNodes()
-    {
-        return shuttingDownNodes;
-    }
-
-    public Set<InternalNode> getDrainedNodes()
-    {
-        return drainedNodes;
-    }
-
-    public Set<InternalNode> getDrainingNodes()
-    {
-        return drainingNodes;
-    }
-
-    public Set<InternalNode> getActiveCoordinators()
-    {
-        return activeCoordinators;
-    }
-
-    @Override
-    public boolean equals(Object o)
-    {
-        if (this == o) {
-            return true;
-        }
-        if (o == null || getClass() != o.getClass()) {
-            return false;
-        }
-        AllNodes allNodes = (AllNodes) o;
-        return Objects.equals(activeNodes, allNodes.activeNodes) &&
-                Objects.equals(inactiveNodes, allNodes.inactiveNodes) &&
-                Objects.equals(drainedNodes, allNodes.drainedNodes) &&
-                Objects.equals(drainingNodes, allNodes.drainingNodes) &&
-                Objects.equals(shuttingDownNodes, allNodes.shuttingDownNodes) &&
-                Objects.equals(activeCoordinators, allNodes.activeCoordinators);
-    }
-
-    @Override
-    public int hashCode()
-    {
-        return Objects.hash(activeNodes, inactiveNodes, drainingNodes, drainedNodes, shuttingDownNodes, activeCoordinators);
+    public AllNodes
+    {
+        activeNodes = ImmutableSet.copyOf(requireNonNull(activeNodes, "activeNodes is null"));
+        inactiveNodes = ImmutableSet.copyOf(requireNonNull(inactiveNodes, "inactiveNodes is null"));
+        drainedNodes = ImmutableSet.copyOf(requireNonNull(drainedNodes, "drainedNodes is null"));
+        drainingNodes = ImmutableSet.copyOf(requireNonNull(drainingNodes, "drainingNodes is null"));
+        shuttingDownNodes = ImmutableSet.copyOf(requireNonNull(shuttingDownNodes, "shuttingDownNodes is null"));
+        activeCoordinators = ImmutableSet.copyOf(requireNonNull(activeCoordinators, "activeCoordinators is null"));
     }
 }
diff --git a/core/trino-main/src/main/java/io/trino/node/CoordinatorNodeManager.java b/core/trino-main/src/main/java/io/trino/node/CoordinatorNodeManager.java
@@ -246,31 +246,31 @@ public synchronized AllNodes getAllNodes()
     @Managed
     public int getActiveNodeCount()
     {
-        return getAllNodes().getActiveNodes().size();
+        return getAllNodes().activeNodes().size();
     }
 
     @Managed
     public int getInactiveNodeCount()
     {
-        return getAllNodes().getInactiveNodes().size();
+        return getAllNodes().inactiveNodes().size();
     }
 
     @Managed
     public int getDrainingNodeCount()
     {
-        return getAllNodes().getDrainingNodes().size();
+        return getAllNodes().drainingNodes().size();
     }
 
     @Managed
     public int getDrainedNodeCount()
     {
-        return getAllNodes().getDrainedNodes().size();
+        return getAllNodes().drainedNodes().size();
     }
 
     @Managed
     public int getShuttingDownNodeCount()
     {
-        return getAllNodes().getShuttingDownNodes().size();
+        return getAllNodes().shuttingDownNodes().size();
     }
 
     @VisibleForTesting
diff --git a/core/trino-main/src/main/java/io/trino/node/InternalNodeManager.java b/core/trino-main/src/main/java/io/trino/node/InternalNodeManager.java
@@ -27,23 +27,23 @@ public interface InternalNodeManager
     default Set<InternalNode> getNodes(NodeState state)
     {
         return switch (state) {
-            case ACTIVE -> getAllNodes().getActiveNodes();
-            case INACTIVE -> getAllNodes().getInactiveNodes();
-            case DRAINING -> getAllNodes().getDrainingNodes();
-            case DRAINED -> getAllNodes().getDrainedNodes();
-            case SHUTTING_DOWN -> getAllNodes().getShuttingDownNodes();
+            case ACTIVE -> getAllNodes().activeNodes();
+            case INACTIVE -> getAllNodes().inactiveNodes();
+            case DRAINING -> getAllNodes().drainingNodes();
+            case DRAINED -> getAllNodes().drainedNodes();
+            case SHUTTING_DOWN -> getAllNodes().shuttingDownNodes();
             case INVALID, GONE -> ImmutableSet.of();
         };
     }
 
     default NodesSnapshot getActiveNodesSnapshot()
     {
-        return new NodesSnapshot(getAllNodes().getActiveNodes());
+        return new NodesSnapshot(getAllNodes().activeNodes());
     }
 
     default Set<InternalNode> getCoordinators()
     {
-        return getAllNodes().getActiveCoordinators();
+        return getAllNodes().activeCoordinators();
     }
 
     AllNodes getAllNodes();
diff --git a/core/trino-main/src/main/java/io/trino/node/TestingInternalNodeManager.java b/core/trino-main/src/main/java/io/trino/node/TestingInternalNodeManager.java
@@ -88,7 +88,7 @@ public synchronized void addNodes(Collection<InternalNode> internalNodes)
     {
         checkArgument(internalNodes.stream().noneMatch(currentNode::equals), "Cannot add current node");
         Set<InternalNode> newActiveNodes = ImmutableSet.<InternalNode>builder()
-                .addAll(allNodes.getActiveNodes())
+                .addAll(allNodes.activeNodes())
                 .addAll(internalNodes)
                 .build();
 
@@ -108,7 +108,7 @@ public synchronized void removeNode(InternalNode internalNode)
         requireNonNull(internalNode, "internalNode is null");
         checkArgument(!currentNode.equals(internalNode), "Cannot remove current node");
 
-        Set<InternalNode> newActiveNodes = new HashSet<>(allNodes.getActiveNodes());
+        Set<InternalNode> newActiveNodes = new HashSet<>(allNodes.activeNodes());
         newActiveNodes.remove(internalNode);
 
         setAllNodes(new AllNodes(
diff --git a/core/trino-main/src/main/java/io/trino/server/testing/TestingTrinoServer.java b/core/trino-main/src/main/java/io/trino/server/testing/TestingTrinoServer.java
@@ -691,7 +691,7 @@ public void unregisterServer(InternalNode worker)
 
     public int getWorkerCount()
     {
-        return (int) nodeManager.getAllNodes().getActiveNodes().stream()
+        return (int) nodeManager.getAllNodes().activeNodes().stream()
                 .filter(node -> !currentNode.equals(node))
                 .count();
     }
diff --git a/core/trino-main/src/main/java/io/trino/server/ui/WorkerResource.java b/core/trino-main/src/main/java/io/trino/server/ui/WorkerResource.java
@@ -124,8 +124,8 @@ public Response getThreads(
     @GET
     public Response getWorkerList()
     {
-        Set<InternalNode> activeNodes = nodeManager.getAllNodes().getActiveNodes();
-        Set<InternalNode> inactiveNodes = nodeManager.getAllNodes().getInactiveNodes();
+        Set<InternalNode> activeNodes = nodeManager.getAllNodes().activeNodes();
+        Set<InternalNode> inactiveNodes = nodeManager.getAllNodes().inactiveNodes();
         Set<JsonNodeInfo> jsonNodes = new HashSet<>();
         for (Node node : activeNodes) {
             JsonNodeInfo jsonNode = new JsonNodeInfo(node.getNodeIdentifier(), node.getHostAndPort().getHostText(), node.getVersion(), node.isCoordinator(), ACTIVE.toString().toLowerCase(Locale.ENGLISH));
diff --git a/core/trino-main/src/test/java/io/trino/node/TestCoordinatorNodeManager.java b/core/trino-main/src/test/java/io/trino/node/TestCoordinatorNodeManager.java
@@ -118,7 +118,7 @@ void testGetAllNodes()
             assertThat(connectorNodes).hasSize(4);
             assertThat(connectorNodes.stream().anyMatch(InternalNode::isCoordinator)).isTrue();
 
-            Set<InternalNode> activeNodes = allNodes.getActiveNodes();
+            Set<InternalNode> activeNodes = allNodes.activeNodes();
             assertThat(activeNodes).containsExactlyInAnyOrderElementsOf(this.activeNodes);
 
             for (InternalNode actual : activeNodes) {
@@ -129,7 +129,7 @@ void testGetAllNodes()
 
             assertThat(activeNodes).containsExactlyInAnyOrderElementsOf(manager.getNodes(ACTIVE));
 
-            Set<InternalNode> inactiveNodes = allNodes.getInactiveNodes();
+            Set<InternalNode> inactiveNodes = allNodes.inactiveNodes();
             assertThat(inactiveNodes).containsExactlyInAnyOrderElementsOf(this.inactiveNodes);
 
             for (InternalNode actual : inactiveNodes) {
@@ -186,17 +186,17 @@ void testNodeChangeListener()
             manager.addNodeChangeListener(notifications::add);
             AllNodes allNodes = notifications.take();
             assertThat(manager.getAllNodes()).isSameAs(allNodes);
-            assertThat(allNodes.getActiveNodes()).containsExactly(currentNode);
-            assertThat(allNodes.getInactiveNodes()).isEmpty();
+            assertThat(allNodes.activeNodes()).containsExactly(currentNode);
+            assertThat(allNodes.inactiveNodes()).isEmpty();
 
             // announce all nodes
             testingTicker.increment(5, SECONDS);
             selector.announceNodes(activeNodes, inactiveNodes);
             manager.refreshNodes();
             allNodes = notifications.take();
             assertThat(manager.getAllNodes()).isSameAs(allNodes);
-            assertThat(allNodes.getActiveNodes()).isEqualTo(activeNodes);
-            assertThat(allNodes.getActiveCoordinators()).isEqualTo(ImmutableSet.of(coordinator));
+            assertThat(allNodes.activeNodes()).isEqualTo(activeNodes);
+            assertThat(allNodes.activeCoordinators()).isEqualTo(ImmutableSet.of(coordinator));
 
             // only announce current node and inactive nodes
             // node manager tracks all nodes until they have not been seen for a while
@@ -205,8 +205,8 @@ void testNodeChangeListener()
             manager.refreshNodes();
             allNodes = notifications.take();
             assertThat(manager.getAllNodes()).isSameAs(allNodes);
-            assertThat(allNodes.getActiveNodes()).containsExactly(currentNode);
-            assertThat(allNodes.getInactiveNodes()).isEqualTo(inactiveNodes);
+            assertThat(allNodes.activeNodes()).containsExactly(currentNode);
+            assertThat(allNodes.inactiveNodes()).isEqualTo(inactiveNodes);
         }
         finally {
             manager.stop();

Original file line number	Diff line number	Diff line change
`@@ -128,7 +128,7 @@ public ThreadPoolExecutorMBean getExecutor()`
`128`	`128`	`public void pruneWorkerCatalogs()`
`129`	`129`	`{`
`130`	`130`	`AllNodes allNodes = internalNodeManager.getAllNodes();`
`131`		`- Set<URI> online = Stream.of(allNodes.getActiveNodes(), allNodes.getInactiveNodes(), allNodes.getDrainingNodes(), allNodes.getDrainedNodes(), allNodes.getShuttingDownNodes())`
	`131`	`+ Set<URI> online = Stream.of(allNodes.activeNodes(), allNodes.inactiveNodes(), allNodes.drainingNodes(), allNodes.drainedNodes(), allNodes.shuttingDownNodes())`
`132`	`132`	`.flatMap(Set::stream)`
`133`	`133`	`.map(InternalNode::getInternalUri)`
`134`	`134`	`.filter(uri -> !uri.equals(currentNode.getInternalUri()))`
Original file line number	Diff line number	Diff line change
`@@ -137,10 +137,10 @@ private synchronized void removeFuture(MinNodesFuture minNodesFuture)`
`137`	`137`	`private synchronized void updateAllNodes(AllNodes allNodes)`
`138`	`138`	`{`
`139`	`139`	`if (includeCoordinator) {`
`140`		`- currentCount = allNodes.getActiveNodes().size();`
	`140`	`+ currentCount = allNodes.activeNodes().size();`
`141`	`141`	`}`
`142`	`142`	`else {`
`143`		`- currentCount = Sets.difference(allNodes.getActiveNodes(), allNodes.getActiveCoordinators()).size();`
	`143`	`+ currentCount = Sets.difference(allNodes.activeNodes(), allNodes.activeCoordinators()).size();`
`144`	`144`	`}`
`145`	`145`
`146`	`146`	`ImmutableList.Builder<SettableFuture<Void>> listenersBuilder = ImmutableList.builder();`
Original file line number	Diff line number	Diff line change
`@@ -246,31 +246,31 @@ public synchronized AllNodes getAllNodes()`
`246`	`246`	`@Managed`
`247`	`247`	`public int getActiveNodeCount()`
`248`	`248`	`{`
`249`		`- return getAllNodes().getActiveNodes().size();`
	`249`	`+ return getAllNodes().activeNodes().size();`
`250`	`250`	`}`
`251`	`251`
`252`	`252`	`@Managed`
`253`	`253`	`public int getInactiveNodeCount()`
`254`	`254`	`{`
`255`		`- return getAllNodes().getInactiveNodes().size();`
	`255`	`+ return getAllNodes().inactiveNodes().size();`
`256`	`256`	`}`
`257`	`257`
`258`	`258`	`@Managed`
`259`	`259`	`public int getDrainingNodeCount()`
`260`	`260`	`{`
`261`		`- return getAllNodes().getDrainingNodes().size();`
	`261`	`+ return getAllNodes().drainingNodes().size();`
`262`	`262`	`}`
`263`	`263`
`264`	`264`	`@Managed`
`265`	`265`	`public int getDrainedNodeCount()`
`266`	`266`	`{`
`267`		`- return getAllNodes().getDrainedNodes().size();`
	`267`	`+ return getAllNodes().drainedNodes().size();`
`268`	`268`	`}`
`269`	`269`
`270`	`270`	`@Managed`
`271`	`271`	`public int getShuttingDownNodeCount()`
`272`	`272`	`{`
`273`		`- return getAllNodes().getShuttingDownNodes().size();`
	`273`	`+ return getAllNodes().shuttingDownNodes().size();`
`274`	`274`	`}`
`275`	`275`
`276`	`276`	`@VisibleForTesting`
Original file line number	Diff line number	Diff line change
`@@ -27,23 +27,23 @@ public interface InternalNodeManager`
`27`	`27`	`default Set<InternalNode> getNodes(NodeState state)`
`28`	`28`	`{`
`29`	`29`	`return switch (state) {`
`30`		`- case ACTIVE -> getAllNodes().getActiveNodes();`
`31`		`- case INACTIVE -> getAllNodes().getInactiveNodes();`
`32`		`- case DRAINING -> getAllNodes().getDrainingNodes();`
`33`		`- case DRAINED -> getAllNodes().getDrainedNodes();`
`34`		`- case SHUTTING_DOWN -> getAllNodes().getShuttingDownNodes();`
	`30`	`+ case ACTIVE -> getAllNodes().activeNodes();`
	`31`	`+ case INACTIVE -> getAllNodes().inactiveNodes();`
	`32`	`+ case DRAINING -> getAllNodes().drainingNodes();`
	`33`	`+ case DRAINED -> getAllNodes().drainedNodes();`
	`34`	`+ case SHUTTING_DOWN -> getAllNodes().shuttingDownNodes();`
`35`	`35`	`case INVALID, GONE -> ImmutableSet.of();`
`36`	`36`	`};`
`37`	`37`	`}`
`38`	`38`
`39`	`39`	`default NodesSnapshot getActiveNodesSnapshot()`
`40`	`40`	`{`
`41`		`- return new NodesSnapshot(getAllNodes().getActiveNodes());`
	`41`	`+ return new NodesSnapshot(getAllNodes().activeNodes());`
`42`	`42`	`}`
`43`	`43`
`44`	`44`	`default Set<InternalNode> getCoordinators()`
`45`	`45`	`{`
`46`		`- return getAllNodes().getActiveCoordinators();`
	`46`	`+ return getAllNodes().activeCoordinators();`
`47`	`47`	`}`
`48`	`48`
`49`	`49`	`AllNodes getAllNodes();`
Original file line number	Diff line number	Diff line change
`@@ -691,7 +691,7 @@ public void unregisterServer(InternalNode worker)`
`691`	`691`
`692`	`692`	`public int getWorkerCount()`
`693`	`693`	`{`
`694`		`- return (int) nodeManager.getAllNodes().getActiveNodes().stream()`
	`694`	`+ return (int) nodeManager.getAllNodes().activeNodes().stream()`
`695`	`695`	`.filter(node -> !currentNode.equals(node))`
`696`	`696`	`.count();`
`697`	`697`	`}`