@Test
    public void shouldNotCreateTopicWhenItAlreadyExists() {
        NewTopic newTopic = TopicAdmin.defineTopic("myTopic").partitions(1).compacted().build();
        Cluster cluster = createCluster(1);
        try (MockAdminClient mockAdminClient = new MockAdminClient(cluster.nodes(), cluster.nodeById(0))) {
            TopicPartitionInfo topicPartitionInfo = new TopicPartitionInfo(0, cluster.nodeById(0), cluster.nodes(), Collections.emptyList());
            mockAdminClient.addTopic(false, "myTopic", Collections.singletonList(topicPartitionInfo), null);
            TopicAdmin admin = new TopicAdmin(mockAdminClient);
            assertFalse(admin.createTopic(newTopic));
            assertTrue(admin.createTopics(newTopic).isEmpty());
            assertTrue(admin.createOrFindTopic(newTopic));
            TopicAdmin.TopicCreationResponse response = admin.createOrFindTopics(newTopic);
            assertTrue(response.isCreatedOrExisting(newTopic.name()));
            assertTrue(response.isExisting(newTopic.name()));
            assertFalse(response.isCreated(newTopic.name()));
        }
    }
0.11.0.0 clients can talk with older brokers, but the CREATE_TOPIC API was added in 0.10.1.0. That means,
if our TopicAdmin talks to a pre 0.10.1 broker, it should receive an UnsupportedVersionException, should
create no topics, and return false.
connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java