@Override
    public CreateAclsResult createAcls(Collection<AclBinding> acls, CreateAclsOptions options) {
        CreateAclsResult aclsResult = super.createAcls(acls, options);
        aclsResult.values().forEach((aclBinding, future) -> future.whenComplete((ignored, error) -> {
            if (error == null) {
                FakeLocalMetadataStore.addACLs(aclBinding.entry().principal(), aclBinding);
            } else {
                log.error("Unable to intercept admin client operation", error);
            }
        }));
        return aclsResult;
    }
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