private PartitionedByError<ReconcilableTopic, Void> createTopics(List<ReconcilableTopic> kts) {
        var newTopics = kts.stream().map(reconcilableTopic -> {
            // Admin create
            return buildNewTopic(reconcilableTopic.kt(), reconcilableTopic.topicName());
        }).collect(Collectors.toSet());

        LOGGER.debugOp("Admin.createTopics({})", newTopics);
        var timerSample = TopicOperatorUtil.startExternalRequestTimer(metrics, enableAdditionalMetrics);
        CreateTopicsResult ctr = admin.createTopics(newTopics);
        ctr.all().whenComplete((i, e) -> {
            TopicOperatorUtil.stopExternalRequestTimer(timerSample, metrics::createTopicsTimer, enableAdditionalMetrics, namespace);
            if (e != null) {
                LOGGER.traceOp("Admin.createTopics({}) failed with {}", newTopics, String.valueOf(e));
            } else {
                LOGGER.traceOp("Admin.createTopics({}) completed", newTopics);
            }
        });
        Map<String, KafkaFuture<Void>> values = ctr.values();
        return partitionedByError(kts.stream().map(reconcilableTopic -> {
            try {
                values.get(reconcilableTopic.topicName()).get();
                reconcilableTopic.kt().setStatus(new KafkaTopicStatusBuilder()
                    .withTopicId(ctr.topicId(reconcilableTopic.topicName()).get().toString()).build());
                return new Pair<>(reconcilableTopic, Either.ofRight((null)));
            } catch (ExecutionException e) {
                if (e.getCause() != null && e.getCause() instanceof TopicExistsException) {
                    // we treat this as a success, the next reconciliation checks the configuration
                    return new Pair<>(reconcilableTopic, Either.ofRight((null)));
                } else {
                    return new Pair<>(reconcilableTopic, Either.ofLeft(handleAdminException(e)));
                }
            } catch (InterruptedException e) {
                throw new UncheckedInterruptedException(e);
            }
        }));
    }
@Test
    public void shouldHandleInterruptedExceptionFromListReassignments(
            @BrokerCluster(numBrokers = 2)
            KafkaCluster cluster) throws ExecutionException, InterruptedException {
        var topicName = "my-topic";
        kafkaAdminClient[0] = Admin.create(Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.getBootstrapServers()));
        kafkaAdminClient[0].createTopics(List.of(new NewTopic(topicName, 2, (short) 2))).all().get();

        var kafkaAdminClientSpy = Mockito.spy(kafkaAdminClient[0]);
        var result = Mockito.mock(ListPartitionReassignmentsResult.class);
        Mockito.doReturn(interruptedFuture()).when(result).reassignments();
        Mockito.doReturn(result).when(kafkaAdminClientSpy).listPartitionReassignments(any(Set.class));
        
        KafkaTopic kafkaTopic = createKafkaTopic(topicName);
        assertOnUpdateThrowsInterruptedException(kafkaAdminClientSpy, kafkaTopic);
    }