public Future<KafkaVersionChange> reconcile()    {
        return getVersionFromController()
                .compose(i -> getPods())
                .compose(this::detectToAndFromVersions)
                .compose(i -> prepareVersionChange());
    }
@Test
    public void testDowngradeFailsWithNewProtocolVersions(VertxTestContext context) {
        String oldKafkaVersion = KafkaVersionTestUtils.LATEST_KAFKA_VERSION;
        String oldInterBrokerProtocolVersion = KafkaVersionTestUtils.LATEST_PROTOCOL_VERSION;
        String oldLogMessageFormatVersion = KafkaVersionTestUtils.LATEST_FORMAT_VERSION;

        String kafkaVersion = KafkaVersionTestUtils.PREVIOUS_KAFKA_VERSION;
        String interBrokerProtocolVersion = KafkaVersionTestUtils.PREVIOUS_PROTOCOL_VERSION;
        String logMessageFormatVersion = KafkaVersionTestUtils.PREVIOUS_FORMAT_VERSION;

        VersionChangeCreator vcc = mockVersionChangeCreator(
                mockKafka(kafkaVersion, interBrokerProtocolVersion, logMessageFormatVersion),
                mockNewCluster(
                        null,
                        mockSps(oldKafkaVersion),
                        mockUniformPods(oldKafkaVersion, oldInterBrokerProtocolVersion, oldLogMessageFormatVersion)
                )
        );

        Checkpoint async = context.checkpoint();
        vcc.reconcile().onComplete(context.failing(c -> context.verify(() -> {
            assertThat(c.getClass(), is(KafkaUpgradeException.class));
            assertThat(c.getMessage(), is("log.message.format.version (" + oldInterBrokerProtocolVersion + ") and inter.broker.protocol.version (" + oldLogMessageFormatVersion + ") used by the brokers have to be set and be lower or equal to the Kafka broker version we downgrade to (" + KafkaVersionTestUtils.PREVIOUS_KAFKA_VERSION + ")"));

            async.flag();
        })));
    }