public static KafkaPool fromCrd(
            Reconciliation reconciliation,
            Kafka kafka,
            KafkaNodePool pool,
            NodeIdAssignment idAssignment,
            Storage oldStorage,
            OwnerReference ownerReference,
            SharedEnvironmentProvider sharedEnvironmentProvider
    ) {
        ModelUtils.validateComputeResources(pool.getSpec().getResources(), "KafkaNodePool.spec.resources");
        StorageUtils.validatePersistentStorage(pool.getSpec().getStorage(), "KafkaNodePool.spec.storage");

        KafkaPool result = new KafkaPool(reconciliation, kafka, pool, componentName(kafka, pool), ownerReference, idAssignment, sharedEnvironmentProvider);

        result.gcLoggingEnabled = isGcLoggingEnabled(kafka, pool);
        result.jvmOptions = pool.getSpec().getJvmOptions() != null ? pool.getSpec().getJvmOptions() : kafka.getSpec().getKafka().getJvmOptions();
        result.resources = pool.getSpec().getResources() != null ? pool.getSpec().getResources() : kafka.getSpec().getKafka().getResources();
        result.processRoles = new HashSet<>(pool.getSpec().getRoles());

        if (oldStorage != null) {
            Storage newStorage = pool.getSpec().getStorage();

            StorageDiff diff = new StorageDiff(reconciliation, oldStorage, newStorage, idAssignment.current(), idAssignment.desired());

            if (diff.issuesDetected()) {
                LOGGER.warnCr(reconciliation, "Only the following changes to Kafka storage are allowed: " +
                        "changing the deleteClaim flag, " +
                        "changing the kraftMetadata flag (but only one one volume can be marked to store the KRaft metadata log at a time), " +
                        "adding volumes to Jbod storage or removing volumes from Jbod storage, " +
                        "each volume in Jbod storage should have an unique ID, " +
                        "changing overrides to nodes which do not exist yet, " +
                        "and increasing size of persistent claim volumes (depending on the volume type and used storage class).");
                LOGGER.warnCr(reconciliation, "The desired Kafka storage configuration in the KafkaNodePool resource {}/{} contains changes which are not allowed. As a " +
                        "result, all storage changes will be ignored. Use DEBUG level logging for more information " +
                        "about the detected changes.", pool.getMetadata().getNamespace(), pool.getMetadata().getName());

                Condition warning = StatusUtils.buildWarningCondition("KafkaStorage",
                        "The desired Kafka storage configuration in the KafkaNodePool resource " + pool.getMetadata().getNamespace() + "/" + pool.getMetadata().getName() + " contains changes which are not allowed. As a " +
                                "result, all storage changes will be ignored. Use DEBUG level logging for more information " +
                                "about the detected changes.");
                result.warningConditions.add(warning);

                result.setStorage(oldStorage);
            } else {
                if (!VolumeUtils.kraftMetadataPath(oldStorage).equals(VolumeUtils.kraftMetadataPath(newStorage)))    {
                    // The volume for the KRaft metadata log is changing. We should log it.
                    LOGGER.warnCr(reconciliation, "The KRaft metadata log for KafkaNodePool {}/{} will be moved from volume {} to volume {}.", pool.getMetadata().getNamespace(), pool.getMetadata().getName(), VolumeUtils.kraftMetadataPath(oldStorage), VolumeUtils.kraftMetadataPath(newStorage));
                }

                result.setStorage(newStorage);
            }
        } else {
            result.setStorage(pool.getSpec().getStorage());
        }

        // Adds the warnings about unknown or deprecated fields
        result.warningConditions.addAll(StatusUtils.validate(reconciliation, pool));

        if (pool.getSpec().getTemplate() != null) {
            KafkaNodePoolTemplate template = pool.getSpec().getTemplate();

            result.templatePersistentVolumeClaims = template.getPersistentVolumeClaim();
            result.templatePodSet = template.getPodSet();
            result.templatePod = template.getPod();
            result.templatePerBrokerService = template.getPerPodService();
            result.templatePerBrokerRoute = template.getPerPodRoute();
            result.templatePerBrokerIngress = template.getPerPodIngress();
            result.templateContainer = template.getKafkaContainer();
            result.templateInitContainer = template.getInitContainer();
        } else if (kafka.getSpec().getKafka().getTemplate() != null) {
            KafkaClusterTemplate template = kafka.getSpec().getKafka().getTemplate();

            result.templatePersistentVolumeClaims = template.getPersistentVolumeClaim();
            result.templatePodSet = template.getPodSet();
            result.templatePod = template.getPod();
            result.templatePerBrokerService = template.getPerPodService();
            result.templatePerBrokerRoute = template.getPerPodRoute();
            result.templatePerBrokerIngress = template.getPerPodIngress();
            result.templateContainer = template.getKafkaContainer();
            result.templateInitContainer = template.getInitContainer();
        }

        return result;
    }
@Test
    public void testStorageValidation()  {
        KafkaNodePool pool = new KafkaNodePoolBuilder(POOL)
                .editSpec()
                    .withNewJbodStorage()
                    .endJbodStorage()
                .endSpec()
                .build();

        InvalidResourceException ex = assertThrows(InvalidResourceException.class, () -> 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()),
                null,
                OWNER_REFERENCE,
                SHARED_ENV_PROVIDER
        ));

        assertThat(ex.getMessage(), containsString("JbodStorage needs to contain at least one volume (KafkaNodePool.spec.storage"));
    }