PartitionedByError<ReconcilableTopic, Void> checkReplicasChanges(List<ReconcilableTopic> reconcilableTopics,
                                                                                PartitionedByError<ReconcilableTopic, TopicState> currentStatesOrError) {
        var differentRfResults = findDifferentRf(currentStatesOrError);
        Stream<Pair<ReconcilableTopic, Either<TopicOperatorException, Void>>> successStream;
            
        if (config.cruiseControlEnabled()) {
            var results = new HashSet<ReconcilableTopic>();
            var differentRfMap = differentRfResults.ok().map(Pair::getKey).collect(Collectors.toList())
                .stream().collect(Collectors.toMap(ReconcilableTopic::topicName, Function.identity()));
            
            var pending = topicsMatching(reconcilableTopics, this::isPendingReplicasChange);
            var brandNew = differentRfMap.values().stream()
                .filter(rt -> !isPendingReplicasChange(rt.kt()) && !isOngoingReplicasChange(rt.kt()))
                .collect(Collectors.toList());
            pending.addAll(brandNew);
            warnTooLargeMinIsr(pending);
            results.addAll(replicasChangeHandler.requestPendingChanges(pending));

            var ongoing = topicsMatching(reconcilableTopics, this::isOngoingReplicasChange);
            results.addAll(replicasChangeHandler.requestOngoingChanges(ongoing));
            
            var completed = pending.stream()
                .filter(rt -> !differentRfMap.containsKey(rt.topicName()) && !isFailedReplicasChange(rt.kt()))
                .collect(Collectors.toList());
            var reverted = pending.stream()
                .filter(rt -> !differentRfMap.containsKey(rt.topicName()) && isFailedReplicasChange(rt.kt()))
                .collect(Collectors.toList());
            completed.addAll(reverted);
            if (!completed.isEmpty()) {
                LOGGER.debugOp("Pending but completed replicas changes, Topics: {}", topicNames(completed));
            }
            completed.forEach(reconcilableTopic -> {
                reconcilableTopic.kt().getStatus().setReplicasChange(null);
            });
            results.addAll(completed);
            
            successStream = results.stream().map(reconcilableTopic -> new Pair<>(reconcilableTopic, Either.ofRight(null)));
            
        } else {
            successStream = differentRfResults.ok().map(pair -> {
                var reconcilableTopic = pair.getKey();
                var specReplicas = replicas(reconcilableTopic.kt());
                var partitions = pair.getValue().partitionsWithDifferentRfThan(specReplicas);
                return new Pair<>(reconcilableTopic, Either.ofLeft(new TopicOperatorException.NotSupported(
                    "Replication factor change not supported, but required for partitions " + partitions)));
            });
        }

        Stream<Pair<ReconcilableTopic, Either<TopicOperatorException, Void>>> errorStream = differentRfResults.errors()
            .map(pair -> new Pair<>(pair.getKey(), Either.ofLeft(pair.getValue())));
        
        return partitionedByError(Stream.concat(successStream, errorStream));
    }
@Test
    public void replicasChangeShouldCompleteWhenSpecIsReverted() {
        var topicName = "my-topic";
        var replicationFactor = 3;

        // setup: pending with error and .spec.replicas == uniqueReplicationFactor
        var config = Mockito.mock(TopicOperatorConfig.class);
        Mockito.doReturn(NAMESPACE).when(config).namespace();
        Mockito.doReturn(true).when(config).useFinalizer();
        Mockito.doReturn(false).when(config).enableAdditionalMetrics();
        Mockito.doReturn(true).when(config).cruiseControlEnabled();

        var describeClusterResult = Mockito.mock(DescribeClusterResult.class);
        Mockito.doReturn(KafkaFuture.completedFuture(List.of())).when(describeClusterResult).nodes();

        var partitionReassignmentResult = Mockito.mock(ListPartitionReassignmentsResult.class);
        var topicPartition = Mockito.mock(TopicPartition.class);
        var partitionReassignment = Mockito.mock(PartitionReassignment.class);
        Mockito.doReturn(KafkaFuture.completedFuture(Map.of(topicPartition, partitionReassignment))).when(partitionReassignmentResult).reassignments();

        var kafkaAdminClient = Mockito.mock(Admin.class);
        Mockito.doReturn(describeClusterResult).when(kafkaAdminClient).describeCluster();
        Mockito.doReturn(partitionReassignmentResult).when(kafkaAdminClient).listPartitionReassignments(any(Set.class));

        var topicDescription = Mockito.mock(TopicDescription.class);
        var topicPartitionInfo = Mockito.mock(TopicPartitionInfo.class);
        Mockito.doReturn(List.of(topicPartitionInfo)).when(topicDescription).partitions();

        var currentState = Mockito.mock(TopicState.class);
        Mockito.doReturn(replicationFactor).when(currentState).uniqueReplicationFactor();
        Mockito.doReturn(topicDescription).when(currentState).description();
        
        var kafkaTopic = new KafkaTopicBuilder()
            .withNewMetadata()
                .withName(topicName)
                .withNamespace(NAMESPACE)
                .addToLabels("key", "VALUE")
            .endMetadata()
            .withNewSpec()
                .withPartitions(25)
                .withReplicas(replicationFactor)
            .endSpec()
            .withStatus(new KafkaTopicStatusBuilder()
                .withReplicasChange(new ReplicasChangeStatusBuilder()
                        .withMessage("Error message")
                        .withState(PENDING)
                        .withTargetReplicas(replicationFactor)
                    .build())
                .build())
            .build();

        var replicasChangeHandler = Mockito.mock(ReplicasChangeHandler.class);
        Mockito.doReturn(List.of()).when(replicasChangeHandler).requestPendingChanges(anyList());
        Mockito.doReturn(List.of()).when(replicasChangeHandler).requestOngoingChanges(anyList());

        var reconcilableTopics = List.of(TopicOperatorTestUtil.reconcilableTopic(kafkaTopic, NAMESPACE));
        PartitionedByError<ReconcilableTopic, TopicState> currentStatesOrError = new PartitionedByError<>(List.of(), List.of());
        
        // run test
        var metricsHolder = new TopicOperatorMetricsHolder(RESOURCE_KIND, null, new TopicOperatorMetricsProvider(new SimpleMeterRegistry()));
        var controller = new BatchingTopicController(config, Map.of("key", "VALUE"), kafkaAdminClient, kubernetesClient, metricsHolder, new ReplicasChangeHandler(config, metricsHolder));
        var results = controller.checkReplicasChanges(reconcilableTopics, currentStatesOrError);

        assertThat(results.ok().count(), is(1L));
        assertThat(results.ok().findFirst().get().getKey().kt().getStatus().getReplicasChange(), is(nullValue()));
    }