private KafkaPool(
            Reconciliation reconciliation,
            Kafka kafka,
            KafkaNodePool pool,
            String componentName,
            OwnerReference ownerReference,
            NodeIdAssignment idAssignment,
            SharedEnvironmentProvider sharedEnvironmentProvider
    ) {
        super(
                reconciliation,
                kafka.getMetadata().getName(),
                kafka.getMetadata().getNamespace(),
                componentName,
                Labels.fromResource(pool)
                        // Strimzi labels
                        .withStrimziKind(kafka.getKind())
                        // This needs ot be selectable through KafkaCluster selector. So we intentionally use the <clusterName>-kafka
                        // as the strimzi.io/name. strimzi.io/pool-name can be used to select through node pool.
                        .withStrimziName(KafkaResources.kafkaComponentName(kafka.getMetadata().getName()))
                        .withStrimziCluster(kafka.getMetadata().getName())
                        .withStrimziComponentType(COMPONENT_TYPE)
                        .withStrimziPoolName(pool.getMetadata().getName())
                        // Kubernetes labels
                        .withKubernetesName(COMPONENT_TYPE)
                        .withKubernetesInstance(kafka.getMetadata().getName())
                        .withKubernetesPartOf(kafka.getMetadata().getName())
                        .withKubernetesManagedBy(STRIMZI_CLUSTER_OPERATOR_NAME),
                ownerReference,
                sharedEnvironmentProvider
        );

        this.poolName = pool.getMetadata().getName();
        this.idAssignment = idAssignment;
    }
@Test
    public void testKafkaPool()  {
        KafkaPool kp = KafkaPool.fromCrd(
                Reconciliation.DUMMY_RECONCILIATION,
                KAFKA,
                POOL,
                new NodeIdAssignment(Set.of(10, 11, 13), Set.of(10, 11, 13), Set.of(), Set.of(), Set.of()),
                new JbodStorageBuilder().withVolumes(new PersistentClaimStorageBuilder().withId(0).withSize("100Gi").build()).build(),
                OWNER_REFERENCE,
                SHARED_ENV_PROVIDER
        );

        assertThat(kp, is(notNullValue()));
        assertThat(kp.componentName, is(CLUSTER_NAME + "-pool"));
        assertThat(kp.processRoles, is(Set.of(ProcessRoles.BROKER)));
        assertThat(kp.isBroker(), is(true));
        assertThat(kp.isController(), is(false));
        assertThat(kp.storage, is(new JbodStorageBuilder().withVolumes(new PersistentClaimStorageBuilder().withId(0).withSize("100Gi").build()).build()));
        assertThat(kp.resources, is(nullValue()));
        assertThat(kp.jvmOptions, is(nullValue()));
        assertThat(kp.gcLoggingEnabled, is(false));
        assertThat(kp.templateContainer, is(nullValue()));
        assertThat(kp.templateInitContainer, is(nullValue()));
        assertThat(kp.templatePod, is(nullValue()));
        assertThat(kp.templatePerBrokerIngress, is(nullValue()));
        assertThat(kp.templatePodSet, is(nullValue()));
        assertThat(kp.templatePerBrokerRoute, is(nullValue()));
        assertThat(kp.templatePerBrokerService, is(nullValue()));
        assertThat(kp.templatePersistentVolumeClaims, is(nullValue()));

        Set<NodeRef> nodes = kp.nodes();
        assertThat(nodes.size(), is(3));
        assertThat(nodes, hasItems(new NodeRef(CLUSTER_NAME + "-pool-10", 10, "pool", false, true),
                new NodeRef(CLUSTER_NAME + "-pool-11", 11, "pool", false, true),
                new NodeRef(CLUSTER_NAME + "-pool-13", 13, "pool", false, true)));

        assertThat(kp.containsNodeId(10), is(true));
        assertThat(kp.containsNodeId(11), is(true));
        assertThat(kp.containsNodeId(12), is(false));
        assertThat(kp.containsNodeId(13), is(true));

        assertThat(kp.nodeRef(11), is(new NodeRef(CLUSTER_NAME + "-pool-11", 11, "pool", false, true)));

        KafkaNodePoolStatus status = kp.generateNodePoolStatus("my-cluster-id");
        assertThat(status.getClusterId(), is("my-cluster-id"));
        assertThat(status.getReplicas(), is(3));
        assertThat(status.getLabelSelector(), is("strimzi.io/cluster=my-cluster,strimzi.io/name=my-cluster-kafka,strimzi.io/kind=Kafka,strimzi.io/pool-name=pool"));
        assertThat(status.getNodeIds().size(), is(3));
        assertThat(status.getNodeIds(), hasItems(10, 11, 13));
        assertThat(status.getRoles().size(), is(1));
        assertThat(status.getRoles(), hasItems(ProcessRoles.BROKER));
    }