static String metadataVersionAtUpgrade(Reconciliation reconciliation, String currentMetadataVersion, KafkaVersion versionFrom)   {
        if (currentMetadataVersion != null) {
            if (compareDottedIVVersions(currentMetadataVersion, versionFrom.metadataVersion()) > 0)    {
                // The current metadata version is newer than the version we are upgrading from
                //     => something went completely wrong, and we should just throw an error
                LOGGER.warnCr(reconciliation, "The current metadata version ({}) has to be lower or equal to the Kafka broker version we upgrade from ({})", currentMetadataVersion, versionFrom.version());
                throw new KafkaUpgradeException("The current metadata version (" + currentMetadataVersion + ") has to be lower or equal to the Kafka broker version we upgrade from (" + versionFrom.version() + ")");
            } else {
                // We stick with the current metadata version for the first phase of the upgrade
                //     => it will be changed in the next phase (next reconciliation)
                LOGGER.infoCr(reconciliation, "The current metadata version {} will be used in the first phase of the upgrade", currentMetadataVersion);
                return currentMetadataVersion;
            }
        } else {
            // Current metadata version is missing. This should normally not happen in upgrade as it suggests
            // we are upgrading without the previous version being properly deployed. But in case it happens,
            // we use the metadata version from the older version first.
            LOGGER.warnCr(reconciliation, "The current metadata version seems to be missing during upgrade which is unexpected. The metadata version {} of the Kafka we are upgrading from will be used.", versionFrom.metadataVersion());
            return versionFrom.metadataVersion();
        }
    }
@Test
    public void testMetadataVersionAtUpgrade() {
        assertThat(KRaftVersionChangeCreator.metadataVersionAtUpgrade(Reconciliation.DUMMY_RECONCILIATION, "3.6-IV2", VERSIONS.defaultVersion()),
                is("3.6-IV2"));

        assertThat(KRaftVersionChangeCreator.metadataVersionAtUpgrade(Reconciliation.DUMMY_RECONCILIATION, "3.4-IV2", VERSIONS.defaultVersion()),
                is("3.4-IV2"));

        assertThat(KRaftVersionChangeCreator.metadataVersionAtUpgrade(Reconciliation.DUMMY_RECONCILIATION, null, VERSIONS.defaultVersion()),
                is(VERSIONS.defaultVersion().metadataVersion()));

        KafkaUpgradeException ex = assertThrows(KafkaUpgradeException.class, () -> KRaftVersionChangeCreator.metadataVersionAtUpgrade(Reconciliation.DUMMY_RECONCILIATION, "5.11-IV2", VERSIONS.defaultVersion()));
        assertThat(ex.getMessage(), is("The current metadata version (5.11-IV2) has to be lower or equal to the Kafka broker version we upgrade from (" + VERSIONS.defaultVersion().version() + ")"));
    }