Repository: apache/kafka
File: server/src/test/java/org/apache/kafka/server/BootstrapControllersIntegrationTest.java (Java)

Code before:
        try (Admin admin = Admin.create(adminConfig(clusterInstance, true))) {
            ListOffsetsResult result = admin.listOffsets(Map.of(new TopicPartition("foo", 0), OffsetSpec.earliest()));
            ExecutionException exception =
                assertThrows(ExecutionException.class,
                    () -> result.all().get(1, TimeUnit.MINUTES));
            assertNotNull(exception.getCause());
            assertEquals(UnsupportedEndpointTypeException.class, exception.getCause().getClass());
            assertEquals("This Admin API is not yet supported when communicating directly with " +
                "the controller quorum.", exception.getCause().getMessage());
        }
    }

    @ClusterTest
    public void testIncrementalAlterConfigsByControllers(ClusterInstance clusterInstance) throws Exception {
        testIncrementalAlterConfigs(clusterInstance, true);
    }

    @ClusterTest
    public void testIncrementalAlterConfigs(ClusterInstance clusterInstance) throws Exception {
        testIncrementalAlterConfigs(clusterInstance, false);
    }

    private void testIncrementalAlterConfigs(ClusterInstance clusterInstance, boolean usingBootstrapControllers) throws Exception {
        Collection<Integer> nodeIds = usingBootstrapControllers ?
                clusterInstance.controllerIds() :
                List.of(clusterInstance.brokers().values().iterator().next().config().nodeId());
        try (Admin admin = Admin.create(adminConfig(clusterInstance, usingBootstrapControllers))) {
            for (int nodeId : nodeIds) {
                ConfigResource nodeResource = new ConfigResource(BROKER, "" + nodeId);
                ConfigResource defaultResource = new ConfigResource(BROKER, "");
                Map<ConfigResource, Collection<AlterConfigOp>> alterations = Map.of(
                        nodeResource, List.of(new AlterConfigOp(new ConfigEntry("my.custom.config", "foo"), AlterConfigOp.OpType.SET)),
                        defaultResource, List.of(new AlterConfigOp(new ConfigEntry("my.custom.config", "bar"), AlterConfigOp.OpType.SET))
                );
                admin.incrementalAlterConfigs(alterations).all().get(1, TimeUnit.MINUTES);
                TestUtils.retryOnExceptionWithTimeout(30_000, () -> {
                    Config config = admin.describeConfigs(List.of(nodeResource)).
                            all().get(1, TimeUnit.MINUTES).get(nodeResource);
                    ConfigEntry entry = config.entries().stream().
                            filter(e -> e.name().equals("my.custom.config")).
                            findFirst().orElseThrow();
                    assertEquals(DYNAMIC_BROKER_CONFIG, entry.source(),
                            "Expected entry for my.custom.config to come from DYNAMIC_BROKER_CONFIG. " +
                                    "Instead, the entry was: " + entry);
                });
            }
        }
    }

    @ClusterTest(brokers = 3)
    public void testAlterReassignmentsWithBootstrapControllers(ClusterInstance clusterInstance) throws ExecutionException, InterruptedException {

Review comment (suggestion):
```java
        Collection<Integer> nodeIds = usingBootstrapControllers ?
                clusterInstance.controllerIds() :
                clusterInstance.brokers().keySet();
```

Code after:

    @ClusterTest
    public void testUsingBootstrapControllersOnUnsupportedAdminApi(ClusterInstance clusterInstance) {
        try (Admin admin = Admin.create(adminConfig(clusterInstance, true))) {
            ListOffsetsResult result = admin.listOffsets(Map.of(new TopicPartition("foo", 0), OffsetSpec.earliest()));
            ExecutionException exception =
                assertThrows(ExecutionException.class,
                    () -> result.all().get(1, TimeUnit.MINUTES));
            assertNotNull(exception.getCause());
            assertEquals(UnsupportedEndpointTypeException.class, exception.getCause().getClass());
            assertEquals("This Admin API is not yet supported when communicating directly with " +
                "the controller quorum.", exception.getCause().getMessage());
        }
    }

    @ClusterTest
    public void testIncrementalAlterConfigsByControllers(ClusterInstance clusterInstance) throws Exception {
        testIncrementalAlterConfigs(clusterInstance, true);
    }

    @ClusterTest
    public void testIncrementalAlterConfigs(ClusterInstance clusterInstance) throws Exception {
        testIncrementalAlterConfigs(clusterInstance, false);
    }

    private void testIncrementalAlterConfigs(ClusterInstance clusterInstance, boolean usingBootstrapControllers) throws Exception {
        Collection<Integer> nodeIds = usingBootstrapControllers ?
                clusterInstance.controllerIds() : clusterInstance.brokers().keySet();
        try (Admin admin = Admin.create(adminConfig(clusterInstance, usingBootstrapControllers))) {
            for (int nodeId : nodeIds) {
                ConfigResource nodeResource = new ConfigResource(BROKER, "" + nodeId);
                ConfigResource defaultResource = new ConfigResource(BROKER, "");
                String nodeMaxConnectionsValue = String.valueOf(1000 + nodeId);
                String defaultMaxConnectionsValue = String.valueOf(2000 + nodeId);
                String defaultConnectionRateValue = String.valueOf(2000 + nodeId);

                // Set configs: MAX_CONNECTIONS_CONFIG for per-broker, both configs for default
                Map<ConfigResource, Collection<AlterConfigOp>> alterations = Map.of(
                        nodeResource, List.of(
                                new AlterConfigOp(new ConfigEntry(SocketServerConfigs.MAX_CONNECTIONS_CONFIG, nodeMaxConnectionsValue), AlterConfigOp.OpType.SET)
                        ),
                        defaultResource, List.of(
                                new AlterConfigOp(new ConfigEntry(SocketServerConfigs.MAX_CONNECTIONS_CONFIG, defaultMaxConnectionsValue), AlterConfigOp.OpType.SET),
                                new AlterConfigOp(new ConfigEntry(SocketServerConfigs.MAX_CONNECTION_CREATION_RATE_CONFIG, defaultConnectionRateValue), AlterConfigOp.OpType.SET)
                        )
                );
                admin.incrementalAlterConfigs(alterations).all().get(1, TimeUnit.MINUTES);
                
                // Verify per-broker configs: MAX_CONNECTIONS_CONFIG and MAX_CONNECTION_CREATION_RATE_CONFIG
                verifyConfigValue(admin, nodeResource, SocketServerConfigs.MAX_CONNECTIONS_CONFIG,
                        DYNAMIC_BROKER_CONFIG, nodeMaxConnectionsValue);