public Future<KafkaVersionChange> reconcile()    {
        return getVersionFromController()
                .compose(i -> getPods())
                .compose(this::detectToAndFromVersions)
                .compose(i -> prepareVersionChange());
    }
@Test
    public void testUpgradeWithIbpv(VertxTestContext context) {
        String kafkaVersion = VERSIONS.defaultVersion().version();
        String oldInterBrokerProtocolVersion = KafkaVersionTestUtils.PREVIOUS_PROTOCOL_VERSION;
        String oldLogMessageFormatVersion = KafkaVersionTestUtils.PREVIOUS_FORMAT_VERSION;

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

        Checkpoint async = context.checkpoint();
        vcc.reconcile().onComplete(context.succeeding(c -> context.verify(() -> {
            assertThat(c.from(), is(VERSIONS.defaultVersion()));
            assertThat(c.to(), is(VERSIONS.defaultVersion()));
            assertThat(c.interBrokerProtocolVersion(), is(nullValue())); // Is null because it is set in the Kafka CR
            assertThat(c.logMessageFormatVersion(), is(oldInterBrokerProtocolVersion)); // Mirrors the inter.broker.protocol.version

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