static String metadataVersionWithoutKafkaVersionChange(Reconciliation reconciliation, String currentMetadataVersion, String desiredMetadataVersion, KafkaVersion version)   {
        if (desiredMetadataVersion != null)  {
            if (compareDottedIVVersions(version.metadataVersion(), desiredMetadataVersion) >= 0)    {
                return desiredMetadataVersion;
            } else if (currentMetadataVersion != null) {
                LOGGER.warnCr(reconciliation, "Desired metadata version ({}) is newer than the used Kafka version ({}). The cluster will continue to use the current metadata version ({}).", desiredMetadataVersion, version.version(), currentMetadataVersion);
                return currentMetadataVersion;
            } else {
                LOGGER.warnCr(reconciliation, "Desired metadata version ({}) is newer than the used Kafka version ({}). Default metadata version for given Kafka version ({}) will be used instead.", desiredMetadataVersion, version.version(), version.metadataVersion());
                return version.metadataVersion();
            }
        } else {
            // No user-set metadata version & no Kafka version change happening => we use the default for the Kafka version
            return version.metadataVersion();
        }
    }
@Test
    public void testMetadataVersionAtNoChange() {
        assertThat(KRaftVersionChangeCreator.metadataVersionWithoutKafkaVersionChange(Reconciliation.DUMMY_RECONCILIATION, "3.4-IV1", VERSIONS.defaultVersion().metadataVersion(), VERSIONS.defaultVersion()),
                is(VERSIONS.defaultVersion().messageVersion()));

        assertThat(KRaftVersionChangeCreator.metadataVersionWithoutKafkaVersionChange(Reconciliation.DUMMY_RECONCILIATION, "3.4-IV1", null, VERSIONS.defaultVersion()),
                is(VERSIONS.defaultVersion().messageVersion()));

        assertThat(KRaftVersionChangeCreator.metadataVersionWithoutKafkaVersionChange(Reconciliation.DUMMY_RECONCILIATION, null, null, VERSIONS.defaultVersion()),
                is(VERSIONS.defaultVersion().messageVersion()));

        assertThat(KRaftVersionChangeCreator.metadataVersionWithoutKafkaVersionChange(Reconciliation.DUMMY_RECONCILIATION, "3.4-IV1", "3.5", VERSIONS.defaultVersion()),
                is("3.5"));

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

        assertThat(KRaftVersionChangeCreator.metadataVersionWithoutKafkaVersionChange(Reconciliation.DUMMY_RECONCILIATION, "3.4-IV1", "5.11", VERSIONS.defaultVersion()),
                is("3.4-IV1"));

        assertThat(KRaftVersionChangeCreator.metadataVersionWithoutKafkaVersionChange(Reconciliation.DUMMY_RECONCILIATION, null, "5.11", VERSIONS.defaultVersion()),
                is(VERSIONS.defaultVersion().messageVersion()));
    }