public static Future<Void> maybeUpdateMetadataVersion(
            Reconciliation reconciliation,
            Vertx vertx,
            TlsPemIdentity coTlsPemIdentity,
            AdminClientProvider adminClientProvider,
            String desiredMetadataVersion,
            KafkaStatus status
    ) {
        String bootstrapHostname = KafkaResources.bootstrapServiceName(reconciliation.name()) + "." + reconciliation.namespace() + ".svc:" + KafkaCluster.REPLICATION_PORT;
        LOGGER.debugCr(reconciliation, "Creating AdminClient for Kafka cluster in namespace {}", reconciliation.namespace());
        Admin kafkaAdmin = adminClientProvider.createAdminClient(bootstrapHostname, coTlsPemIdentity.pemTrustSet(), coTlsPemIdentity.pemAuthIdentity());

        Promise<Void> updatePromise = Promise.promise();
        maybeUpdateMetadataVersion(reconciliation, vertx, kafkaAdmin, desiredMetadataVersion, status)
                .onComplete(res -> {
                    // Close the Admin client and return the original result
                    LOGGER.debugCr(reconciliation, "Closing the Kafka Admin API connection");
                    kafkaAdmin.close();
                    updatePromise.handle(res);
                });

        return updatePromise.future();
    }
@Test
    public void testUnsuccessfulMetadataVersionChange(VertxTestContext context)   {
        // Mock the Admin client
        Admin mockAdminClient = mock(Admin.class);

        // Mock describing the current metadata version
        mockDescribeVersion(mockAdminClient);

        // Mock updating metadata version
        @SuppressWarnings(value = "unchecked")
        KafkaFuture<Void> kf = mock(KafkaFuture.class);
        when(kf.whenComplete(any())).thenAnswer(i -> {
            KafkaFuture.BiConsumer<Void, Throwable> action = i.getArgument(0);
            action.accept(null, new InvalidUpdateVersionException("Test error ..."));
            return null;
        });
        UpdateFeaturesResult ufr = mock(UpdateFeaturesResult.class);
        when(ufr.values()).thenReturn(Map.of(KRaftMetadataManager.METADATA_VERSION_KEY, kf));
        when(mockAdminClient.updateFeatures(any(), any())).thenReturn(ufr);

        // Mock the Admin client provider
        AdminClientProvider mockAdminClientProvider = mockAdminClientProvider(mockAdminClient);

        // Dummy KafkaStatus to check the values from
        KafkaStatus status = new KafkaStatus();

        Checkpoint checkpoint = context.checkpoint();
        KRaftMetadataManager.maybeUpdateMetadataVersion(Reconciliation.DUMMY_RECONCILIATION, vertx, DUMMY_IDENTITY, mockAdminClientProvider, "3.6", status)
                .onComplete(context.succeeding(s -> {
                    assertThat(status.getKafkaMetadataVersion(), is("3.6-IV1"));
                    assertThat(status.getConditions().size(), is(1));
                    assertThat(status.getConditions().get(0).getType(), is("Warning"));
                    assertThat(status.getConditions().get(0).getReason(), is("MetadataUpdateFailed"));
                    assertThat(status.getConditions().get(0).getMessage(), is("Failed to update metadata version to 3.6"));

                    verify(mockAdminClient, times(1)).updateFeatures(any(), any());
                    verify(mockAdminClient, times(2)).describeFeatures();

                    checkpoint.flag();
                }));
    }