static void createCompactedTopic(String topicName, short partitions, short replicationFactor, Admin admin) {
        NewTopic topicDescription = TopicAdmin.defineTopic(topicName).
                compacted().
                partitions(partitions).
                replicationFactor(replicationFactor).
                build();

        CreateTopicsOptions args = new CreateTopicsOptions().validateOnly(false);
        try {
            admin.createTopics(singleton(topicDescription), args).values().get(topicName).get();
            log.info("Created topic '{}'", topicName);
        } catch (InterruptedException e) {
            Thread.interrupted();
            throw new ConnectException("Interrupted while attempting to create/find topic '" + topicName + "'", e);
        } catch (ExecutionException e) {
            Throwable cause = e.getCause();
            if (cause instanceof TopicExistsException) {
                log.debug("Unable to create topic '{}' since it already exists.", topicName);
                return;
            }
            if (cause instanceof UnsupportedVersionException) {
                log.debug("Unable to create topic '{}' since the brokers do not support the CreateTopics API." +
                                " Falling back to assume topic exists or will be auto-created by the broker.",
                        topicName);
                return;
            }
            if (cause instanceof TopicAuthorizationException) {
                log.debug("Not authorized to create topic(s) '{}' upon the brokers." +
                                " Falling back to assume topic(s) exist or will be auto-created by the broker.",
                        topicName);
                return;
            }
            if (cause instanceof ClusterAuthorizationException) {
                log.debug("Not authorized to create topic '{}'." +
                                " Falling back to assume topic exists or will be auto-created by the broker.",
                        topicName);
                return;
            }
            if (cause instanceof InvalidConfigurationException) {
                throw new ConnectException("Unable to create topic '" + topicName + "': " + cause.getMessage(),
                        cause);
            }
            if (cause instanceof TimeoutException) {
                // Timed out waiting for the operation to complete
                throw new ConnectException("Timed out while checking for or creating topic '" + topicName + "'." +
                        " This could indicate a connectivity issue, unavailable topic partitions, or if" +
                        " this is your first use of the topic it may have taken too long to create.", cause);
            }
            throw new ConnectException("Error while attempting to create/find topic '" + topicName + "'", e);
        }

    }
@Test
    public void testCreateCompactedTopic() throws Exception {
        Map<String, KafkaFuture<Void>> values = Collections.singletonMap(TOPIC, future);
        when(future.get()).thenReturn(null);
        when(ctr.values()).thenReturn(values);
        when(admin.createTopics(any(), any())).thenReturn(ctr);
        MirrorUtils.createCompactedTopic(TOPIC, (short) 1, (short) 1, admin);

        verify(future).get();
        verify(ctr).values();
        verify(admin).createTopics(any(), any());
    }