@Override
    public CreateTopicsResult createTopics(Collection<NewTopic> newTopics, CreateTopicsOptions options) {
        CreateTopicsResult createTopicsResult = super.createTopics(newTopics, options);
        newTopics.forEach(newTopic -> createTopicsResult.values().get(newTopic.name()).whenComplete((ignored, error) -> {
            if (error == null) {
                FakeLocalMetadataStore.addTopicToLocalMetadataStore(newTopic);
            } else if (error.getCause() instanceof TopicExistsException) {
                log.warn("Topic '{}' already exists. Update the local metadata store if absent", newTopic.name());
                FakeLocalMetadataStore.addTopicToLocalMetadataStore(newTopic);
            } else {
                log.error("Unable to intercept admin client operation", error);
            }
        }));
        return createTopicsResult;
    }
Customised ForwardingAdmin for testing only.
The class create/alter topics, partitions and ACLs in Kafka then store metadata in {@link FakeLocalMetadataStore}.
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/clients/admin/FakeForwardingAdminWithLocalMetadata.java