public static KafkaNodePool convertKafkaToVirtualNodePool(Kafka kafka, Integer existingReplicas) {
        List<Integer> nodeIds = null;

        if (existingReplicas != null && existingReplicas > 0)   {
            // We have at least one existing replica => we prepare the existing node IDs accordingly
            nodeIds = IntStream.range(0, existingReplicas).boxed().toList();
        }

        return new KafkaNodePoolBuilder()
                .withNewMetadata()
                    .withName(DEFAULT_NODE_POOL_NAME)
                    .withNamespace(kafka.getMetadata().getNamespace())
                    .withLabels(kafka.getMetadata().getLabels())
                .endMetadata()
                .withNewSpec()
                    .withReplicas(kafka.getSpec().getKafka().getReplicas())
                    .withStorage(kafka.getSpec().getKafka().getStorage())
                    .withRoles(List.of(ProcessRoles.BROKER)) // We do not need to care about the controller role here since this is only used with ZooKeeper based clusters
                    .withResources(kafka.getSpec().getKafka().getResources())
                    .withJvmOptions(kafka.getSpec().getKafka().getJvmOptions())
                    .withTemplate(convertTemplate(kafka.getSpec().getKafka().getTemplate()))
                .endSpec()
                .withNewStatus()
                    .withNodeIds(nodeIds)
                    .withRoles(ProcessRoles.BROKER)
                .endStatus()
                .build();
    }
@Test
    public void testConvertMaximalKafka()  {
        Kafka kafka = new KafkaBuilder()
                .withNewMetadata()
                    .withName("my-cluster")
                    .withNamespace("my-namespace")
                    .withLabels(Map.of("custom-label", "custom-label-value"))
                    .withAnnotations(Map.of("custom-anno", "custom-anno-value"))
                .endMetadata()
                .withNewSpec()
                    .withNewKafka()
                        .withReplicas(3)
                        .withNewJbodStorage()
                            .withVolumes(new PersistentClaimStorageBuilder().withId(0).withSize("100Gi").build())
                        .endJbodStorage()
                        .withResources(new ResourceRequirementsBuilder().withRequests(Map.of("cpu", new Quantity("4"), "memory", new Quantity("16Gi"))).build())
                        .withNewJvmOptions()
                            .withXms("2048m")
                            .withXmx("4096m")
                        .endJvmOptions()
                        .withNewTemplate()
                            .withNewKafkaContainer()
                                .addToEnv(new ContainerEnvVarBuilder().withName("MY_ENV_VAR").withValue("my-env-var-value").build())
                            .endKafkaContainer()
                        .endTemplate()
                    .endKafka()
                .endSpec()
                .build();

        KafkaNodePool pool = VirtualNodePoolConverter.convertKafkaToVirtualNodePool(kafka, 3);

        // Metadata
        assertThat(pool.getMetadata().getName(), is(VirtualNodePoolConverter.DEFAULT_NODE_POOL_NAME));
        assertThat(pool.getMetadata().getNamespace(), is("my-namespace"));
        assertThat(pool.getMetadata().getLabels(), is(Map.of("custom-label", "custom-label-value")));
        assertThat(pool.getMetadata().getAnnotations().size(), is(0));

        // Spec
        assertThat(pool.getSpec().getReplicas(), is(3));

        JbodStorage storage = (JbodStorage) pool.getSpec().getStorage();
        assertThat(storage.getVolumes().size(), is(1));
        assertThat(storage.getVolumes().get(0).getId(), is(0));
        assertThat(((PersistentClaimStorage) storage.getVolumes().get(0)).getSize(), is("100Gi"));

        assertThat(pool.getSpec().getRoles(), is(List.of(ProcessRoles.BROKER)));

        assertThat(pool.getSpec().getResources().getRequests(), is(Map.of("cpu", new Quantity("4"), "memory", new Quantity("16Gi"))));
        assertThat(pool.getSpec().getJvmOptions().getXms(), is("2048m"));
        assertThat(pool.getSpec().getJvmOptions().getXmx(), is("4096m"));

        assertThat(pool.getSpec().getTemplate().getKafkaContainer(), is(notNullValue()));
        assertThat(pool.getSpec().getTemplate().getKafkaContainer().getEnv().size(), is(1));
        assertThat(pool.getSpec().getTemplate().getKafkaContainer().getEnv().get(0).getName(), is("MY_ENV_VAR"));
        assertThat(pool.getSpec().getTemplate().getKafkaContainer().getEnv().get(0).getValue(), is("my-env-var-value"));

        // Status
        assertThat(pool.getStatus().getNodeIds().size(), is(3));
        assertThat(pool.getStatus().getNodeIds(), hasItems(0, 1, 2));
        assertThat(pool.getStatus().getRoles().size(), is(1));
        assertThat(pool.getStatus().getRoles(), hasItems(ProcessRoles.BROKER));
    }