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

Code before:
            new CreateTopicsOptions().validateOnly(validateOnly)
        ).all().get();
    }

    @ClusterTest
    public void testValidCreateTopicsRequests(ClusterInstance cluster) throws Exception {
        try (Admin admin = cluster.admin()) {
            cluster.createTopic("topic1", 5, (short) 1);

            validateValidCreateTopicsRequests(
                new NewTopic("topic2", 5, (short) 3),
                admin,
                true
            );

            validateValidCreateTopicsRequests(
                new NewTopic("topic3", 11, (short) 2)
                    .configs(Map.of(TopicConfig.RETENTION_MS_CONFIG, "4999")),
                admin,
                true
            );

            validateValidCreateTopicsRequests(
                new NewTopic("topic4", Map.of(
                    0, Arrays.asList(1, 0),
                    1, Arrays.asList(0, 1)
                )),
                admin,
                false
            );
        }
    }

    private void validateErrorCreateTopicsRequests(NewTopic topic, Admin admin, boolean validateOnly, Class<? extends Throwable> expectedExceptionClass, String expectedErrorMessage) {
        ExecutionException exception = assertThrows(ExecutionException.class, () ->
                admin.createTopics(List.of(topic), new CreateTopicsOptions().validateOnly(validateOnly)).all().get());
        assertInstanceOf(expectedExceptionClass, exception.getCause());
        assertTrue(exception.getMessage().contains(expectedErrorMessage));
    }

    @ClusterTest
    public void testErrorCreateTopicsRequests(ClusterInstance cluster) throws Exception {
        try (Admin admin = cluster.admin()) {
            String existingTopic = "existing-topic";
            cluster.createTopic(existingTopic, 5, (short) 1);

            // Policy violations
            validateErrorCreateTopicsRequests(
                new NewTopic("policy-topic1", 4, (short) 1),
                admin,
                false,

Review comment (suggestion):
```suggestion
                    0, List.of(1, 0),
                    1, List.of(0, 1)
```

Code after:
        ).all().get();
    }

    @ClusterTest
    public void testValidCreateTopicsRequests(ClusterInstance cluster) throws Exception {
        try (Admin admin = cluster.admin()) {
            cluster.createTopic("topic1", 5, (short) 1);

            validateValidCreateTopicsRequests(
                new NewTopic("topic2", 5, (short) 3),
                admin,
                true
            );

            validateValidCreateTopicsRequests(
                new NewTopic("topic3", 11, (short) 2)
                    .configs(Map.of(TopicConfig.RETENTION_MS_CONFIG, "4999")),
                admin,
                true
            );

            validateValidCreateTopicsRequests(
                new NewTopic("topic4", Map.of(
                    0, List.of(1, 0),
                    1, List.of(0, 1)
                )),
                admin,
                false
            );
        }
    }

    private void validateErrorCreateTopicsRequests(NewTopic topic, Admin admin, boolean validateOnly, Class<? extends Throwable> expectedExceptionClass, String expectedErrorMessage) {
        ExecutionException exception = assertThrows(ExecutionException.class, () ->
                admin.createTopics(List.of(topic), new CreateTopicsOptions().validateOnly(validateOnly)).all().get());
        assertEquals(
                expectedExceptionClass,
                exception.getCause().getClass(),
                "Expected " + expectedExceptionClass.getSimpleName() + ", but got " + exception.getCause().getClass().getSimpleName()
        );
        assertTrue(exception.getMessage().contains(expectedErrorMessage));
    }

    @ClusterTest
    public void testErrorCreateTopicsRequests(ClusterInstance cluster) throws Exception {
        try (Admin admin = cluster.admin()) {
            String existingTopic = "existing-topic";
            cluster.createTopic(existingTopic, 5, (short) 1);

            // Policy violations
            validateErrorCreateTopicsRequests(