Repository: apache/kafka
File: clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/AdminClientWithPoliciesIntegrationTest.java (Java)

Code before:
        assertFutureThrows(InvalidConfigurationException.class, alterResult.values().get(topicResource3));
        assertFutureThrows(InvalidRequestException.class, alterResult.values().get(brokerResource));
        assertTrue(validationsForResource(brokerResource).isEmpty(),
            "Should not see the broker resource in the AlterConfig policy when the broker configs are not being updated.");
        validations.clear();

        // Verify that no resources are updated since validate_only = true
        clusterInstance.ensureConsistentMetadata();
        describeResult = adminClient.describeConfigs(List.of(topicResource1, topicResource2, topicResource3, brokerResource));
        configs = describeResult.all().get();
        assertEquals(4, configs.size());

        assertEquals(String.valueOf(LogConfig.DEFAULT_MIN_CLEANABLE_DIRTY_RATIO), configs.get(topicResource1).get(TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG).value());
        assertEquals(String.valueOf(ServerLogConfigs.MIN_IN_SYNC_REPLICAS_DEFAULT), configs.get(topicResource1).get(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG).value());

        assertEquals("0.8", configs.get(topicResource2).get(TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG).value());

        assertNull(configs.get(brokerResource).get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG).value());

        // Do an incremental alter config on the broker, ensure we don't see the broker config we set earlier in the policy
        alterResult = adminClient.incrementalAlterConfigs(Map.of(
            brokerResource, List.of(new AlterConfigOp(new ConfigEntry(SocketServerConfigs.MAX_CONNECTIONS_CONFIG, "9999"), OpType.SET))
        ));
        alterResult.all().get();
        assertEquals(Set.of(SocketServerConfigs.MAX_CONNECTIONS_CONFIG), validationsForResource(brokerResource).get(0).configs().keySet());
    }

    private static List<AlterConfigPolicy.RequestMetadata> validationsForResource(ConfigResource resource) {
        return validations.stream().filter(req -> req.resource().equals(resource)).toList();
    }

    public static class Policy implements AlterConfigPolicy {
        private Map<String, ?> configs;
        private boolean closed = false;


        @Override
        public void configure(Map<String, ?> configs) {
            validations.clear();
            this.configs = configs;
        }

        @Override
        public void validate(AlterConfigPolicy.RequestMetadata requestMetadata) {
            validations.add(requestMetadata);
            assertFalse(closed, "Policy should not be closed");
            assertFalse(configs.isEmpty(), "configure should have been called with non empty configs");
            assertFalse(requestMetadata.configs().isEmpty(), "request configs should not be empty");
            assertFalse(requestMetadata.resource().name().isEmpty(), "resource name should not be empty");
            if (requestMetadata.configs().containsKey(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG)) {
                throw new PolicyViolationException("Min in sync replicas cannot be updated");

Review comment (suggestion):
We use this class by setting the cluster config, but it will be marked as unused by the IDE.
Please add a comment to it.

Code after:
            assertEquals(4, configs.size());

            assertEquals(String.valueOf(LogConfig.DEFAULT_MIN_CLEANABLE_DIRTY_RATIO), configs.get(topicResource1).get(TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG).value());
            assertEquals(String.valueOf(ServerLogConfigs.MIN_IN_SYNC_REPLICAS_DEFAULT), configs.get(topicResource1).get(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG).value());

            assertEquals("0.8", configs.get(topicResource2).get(TopicConfig.MIN_CLEANABLE_DIRTY_RATIO_CONFIG).value());

            assertNull(configs.get(brokerResource).get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG).value());

            // Do an incremental alter config on the broker, ensure we don't see the broker config we set earlier in the policy
            alterResult = adminClient.incrementalAlterConfigs(Map.of(
                    brokerResource, List.of(new AlterConfigOp(new ConfigEntry(SocketServerConfigs.MAX_CONNECTIONS_CONFIG, "9999"), OpType.SET))
            ));
            alterResult.all().get();
            assertEquals(Set.of(SocketServerConfigs.MAX_CONNECTIONS_CONFIG), validationsForResource(brokerResource).get(0).configs().keySet());
        }
    }

    private static List<AlterConfigPolicy.RequestMetadata> validationsForResource(ConfigResource resource) {
        return validations.stream().filter(req -> req.resource().equals(resource)).toList();
    }

    /**
     * Used in @ClusterTestDefaults serverProperties, so it may appear unused in the IDE.
     */
    public static class Policy implements AlterConfigPolicy {
        private Map<String, ?> configs;
        private boolean closed = false;


        @Override
        public void configure(Map<String, ?> configs) {
            validations.clear();
            this.configs = configs;
        }

        @Override
        public void validate(AlterConfigPolicy.RequestMetadata requestMetadata) {
            validations.add(requestMetadata);
            assertFalse(closed, "Policy should not be closed");
            assertFalse(configs.isEmpty(), "configure should have been called with non empty configs");
            assertFalse(requestMetadata.configs().isEmpty(), "request configs should not be empty");
            assertFalse(requestMetadata.resource().name().isEmpty(), "resource name should not be empty");
            if (requestMetadata.configs().containsKey(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG)) {
                throw new PolicyViolationException("Min in sync replicas cannot be updated");
            }
        }

        @Override
        public void close() throws Exception {
            this.closed = true;