static TreeSet<Integer> brokerNodesBecomingControllerOnlyNodes(KafkaNodePool pool, Set<Integer> current, Set<Integer> desired) {
        if (pool.getStatus() != null
                && pool.getSpec().getRoles() != null
                && pool.getStatus().getRoles() != null
                && pool.getStatus().getRoles().contains(ProcessRoles.BROKER) // Used to have the broker role
                && !pool.getSpec().getRoles().contains(ProcessRoles.BROKER)) { // But should not have it anymore
            // Collect all node IDs that are both current and desired (i.e. we do not care about old nodes being scaled down or new nodes being scaled up)
            TreeSet<Integer> usedToBeBroker = new TreeSet<>(desired);
            usedToBeBroker.retainAll(current);
            return usedToBeBroker;
        } else {
            return new TreeSet<>();
        }
    }
@Test
    public void testBrokerRoleRemoval() {
        // New node pool
        KafkaNodePool pool = createPool("pool", 3, Map.of());
        assertThat(NodeIdAssignor.brokerNodesBecomingControllerOnlyNodes(pool, Set.of(), Set.of(0, 1, 2)).size(), is(0));

        // Existing node pool without role change
        pool = createPool("pool", 3, Map.of(), List.of(0, 1, 2), List.of(ProcessRoles.BROKER), List.of(ProcessRoles.BROKER));
        assertThat(NodeIdAssignor.brokerNodesBecomingControllerOnlyNodes(pool, Set.of(0, 1, 2), Set.of(0, 1, 2)).size(), is(0));

        // Existing node pool with role change that keeps broker role
        pool = createPool("pool", 3, Map.of(), List.of(0, 1, 2), List.of(ProcessRoles.BROKER), List.of(ProcessRoles.BROKER, ProcessRoles.CONTROLLER));
        assertThat(NodeIdAssignor.brokerNodesBecomingControllerOnlyNodes(pool, Set.of(0, 1, 2), Set.of(0, 1, 2)).size(), is(0));

        // Existing node pool with role change that removes broker role
        pool = createPool("pool", 3, Map.of(), List.of(0, 1, 2), List.of(ProcessRoles.CONTROLLER), List.of(ProcessRoles.BROKER, ProcessRoles.CONTROLLER));
        assertThat(NodeIdAssignor.brokerNodesBecomingControllerOnlyNodes(pool, Set.of(0, 1, 2), Set.of(0, 1, 2)).size(), is(3));
        assertThat(NodeIdAssignor.brokerNodesBecomingControllerOnlyNodes(pool, Set.of(0, 1, 2), Set.of(0, 1, 2)), hasItems(0, 1, 2));

        // Existing node pool with role change that removes broker role and scale-up
        pool = createPool("pool", 4, Map.of(), List.of(0, 1, 2), List.of(ProcessRoles.CONTROLLER), List.of(ProcessRoles.BROKER, ProcessRoles.CONTROLLER));
        assertThat(NodeIdAssignor.brokerNodesBecomingControllerOnlyNodes(pool, Set.of(0, 1, 2), Set.of(0, 1, 2, 4)).size(), is(3));
        assertThat(NodeIdAssignor.brokerNodesBecomingControllerOnlyNodes(pool, Set.of(0, 1, 2), Set.of(0, 1, 2, 4)), hasItems(0, 1, 2));

        // Existing node pool with role change that removes broker role and scale-down
        pool = createPool("pool", 2, Map.of(), List.of(0, 1, 2), List.of(ProcessRoles.CONTROLLER), List.of(ProcessRoles.BROKER, ProcessRoles.CONTROLLER));
        assertThat(NodeIdAssignor.brokerNodesBecomingControllerOnlyNodes(pool, Set.of(0, 1, 2), Set.of(0, 1)).size(), is(2));
        assertThat(NodeIdAssignor.brokerNodesBecomingControllerOnlyNodes(pool, Set.of(0, 1, 2), Set.of(0, 1)), hasItems(0, 1));

        // Existing node pool without roles in status => might be some legacy cluster before we added them
        pool = createPool("pool", 3, Map.of(), List.of(0, 1, 2), List.of(ProcessRoles.CONTROLLER), null);
        assertThat(NodeIdAssignor.brokerNodesBecomingControllerOnlyNodes(pool, Set.of(0, 1, 2), Set.of(0, 1, 2)).size(), is(0));
    }