Repository: apache/kafka
File: test-common/test-common-runtime/src/test/java/org/apache/kafka/common/test/KafkaClusterTestKitTest.java (Java)

Code before:
                        .setNumControllerNodes(1)
                        .setPerServerProperties(perServerProperties)
                        .build())
        );
        assertEquals("Unknown server id 100, 200 in perServerProperties, the existent server ids are 0, 3000", e.getMessage());
    }

    @ParameterizedTest
    @ValueSource(booleans = {true, false})
    public void testCreateClusterAndCloseWithMultipleLogDirs(boolean combined) throws Exception {
        try (KafkaClusterTestKit cluster = new KafkaClusterTestKit.Builder(
                new TestKitNodes.Builder().
                        setNumBrokerNodes(5).
                        setNumDisksPerBroker(2).
                        setCombined(combined).
                        setNumControllerNodes(3).build()).build()) {

            TestKitNodes nodes = cluster.nodes();
            assertEquals(5, nodes.brokerNodes().size());
            assertEquals(3, nodes.controllerNodes().size());

            nodes.brokerNodes().forEach((brokerId, node) -> {
                assertEquals(2, node.logDataDirectories().size());
                Set<String> expected = new HashSet<>(List.of(String.format("broker_%d_data0", brokerId), String.format("broker_%d_data1", brokerId)));
                if (nodes.isCombined(node.id())) {
                    expected = new HashSet<>(List.of(String.format("combined_%d_0", brokerId), String.format("combined_%d_1", brokerId)));
                }
                assertEquals(
                        expected,
                        node.logDataDirectories().stream()
                                .map(p -> Paths.get(p).getFileName().toString())
                                .collect(Collectors.toSet())
                );
            });

            nodes.controllerNodes().forEach((controllerId, node) -> {
                String expected = combined ? String.format("combined_%d_0", controllerId) : String.format("controller_%d", controllerId);
                assertEquals(expected, Paths.get(node.metadataDirectory()).getFileName().toString());
            });
        }
    }

    @Test
    public void testCreateClusterWithSpecificBaseDir() throws Exception {
        Path baseDirectory = TestUtils.tempDirectory().toPath();
        try (KafkaClusterTestKit cluster = new KafkaClusterTestKit.Builder(
                new TestKitNodes.Builder().
                        setBaseDirectory(baseDirectory).
                        setNumBrokerNodes(1).
                        setCombined(true).
                        setNumControllerNodes(1).build()).build()) {

Review comment (suggestion):
```suggestion
                    expected = Set.of(String.format("combined_%d_0", brokerId), String.format("combined_%d_1", brokerId));
```

Code after:
                .build())
        );
        assertEquals("Unknown server id 100, 200 in perServerProperties, the existent server ids are 0, 3000", e.getMessage());
    }

    @ParameterizedTest
    @ValueSource(booleans = {true, false})
    public void testCreateClusterAndCloseWithMultipleLogDirs(boolean combined) throws Exception {
        try (KafkaClusterTestKit cluster = new KafkaClusterTestKit.Builder(
            new TestKitNodes.Builder().
                setNumBrokerNodes(5).
                setNumDisksPerBroker(2).
                setCombined(combined).
                setNumControllerNodes(3).build()).build()) {

            TestKitNodes nodes = cluster.nodes();
            assertEquals(5, nodes.brokerNodes().size());
            assertEquals(3, nodes.controllerNodes().size());

            nodes.brokerNodes().forEach((brokerId, node) -> {
                assertEquals(2, node.logDataDirectories().size());
                Set<String> expected = Set.of(String.format("broker_%d_data0", brokerId), String.format("broker_%d_data1", brokerId));
                if (nodes.isCombined(node.id())) {
                    expected = Set.of(String.format("combined_%d_0", brokerId), String.format("combined_%d_1", brokerId));
                }
                assertEquals(
                    expected,
                    node.logDataDirectories().stream()
                        .map(p -> Paths.get(p).getFileName().toString())
                        .collect(Collectors.toSet())
                );
            });

            nodes.controllerNodes().forEach((controllerId, node) -> {
                String expected = combined ? String.format("combined_%d_0", controllerId) : String.format("controller_%d", controllerId);
                assertEquals(expected, Paths.get(node.metadataDirectory()).getFileName().toString());
            });
        }
    }

    @Test
    public void testCreateClusterWithSpecificBaseDir() throws Exception {
        Path baseDirectory = TestUtils.tempDirectory().toPath();
        try (KafkaClusterTestKit cluster = new KafkaClusterTestKit.Builder(
            new TestKitNodes.Builder().
                setBaseDirectory(baseDirectory).
                setNumBrokerNodes(1).
                setCombined(true).
                setNumControllerNodes(1).build()).build()) {
            assertEquals(cluster.nodes().baseDirectory(), baseDirectory.toFile().getAbsolutePath());
            cluster.nodes().controllerNodes().values().forEach(controller ->