@Override
    public CreatePartitionsResult createPartitions(Map<String, NewPartitions> newPartitions, CreatePartitionsOptions options) {
        CreatePartitionsResult createPartitionsResult = super.createPartitions(newPartitions, options);
        newPartitions.forEach((topic, newPartition) -> createPartitionsResult.values().get(topic).whenComplete((ignored, error) -> {
            if (error == null) {
                FakeLocalMetadataStore.updatePartitionCount(topic, newPartition.totalCount());
            } else {
                log.error("Unable to intercept admin client operation", error);
            }
        }));
        return createPartitionsResult;
    }
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