void handleAclsSnapshot(AclsImage image, KRaftMigrationOperationConsumer operationConsumer) {
        // Need to compare contents of image with all ACLs in ZK and issue updates
        Map<ResourcePattern, Set<AccessControlEntry>> allAclsInSnapshot = new HashMap<>();

        image.acls().values().forEach(standardAcl -> {
            ResourcePattern resourcePattern = resourcePatternFromAcl(standardAcl);
            allAclsInSnapshot.computeIfAbsent(resourcePattern, __ -> new HashSet<>()).add(
                new AccessControlEntry(standardAcl.principal(), standardAcl.host(), standardAcl.operation(), standardAcl.permissionType())
            );
        });

        Set<ResourcePattern> newResources = new HashSet<>(allAclsInSnapshot.keySet());
        Set<ResourcePattern> resourcesToDelete = new HashSet<>();
        Map<ResourcePattern, Set<AccessControlEntry>> changedResources = new HashMap<>();
        migrationClient.aclClient().iterateAcls((resourcePattern, accessControlEntries) -> {
            newResources.remove(resourcePattern);
            if (!allAclsInSnapshot.containsKey(resourcePattern)) {
                resourcesToDelete.add(resourcePattern);
            } else {
                Set<AccessControlEntry> snapshotEntries = allAclsInSnapshot.get(resourcePattern);
                if (!snapshotEntries.equals(accessControlEntries)) {
                    changedResources.put(resourcePattern, snapshotEntries);
                }
            }
        });

        newResources.forEach(resourcePattern -> {
            // newResources is generated from allAclsInSnapshot, and we don't remove from that map, so this unguarded .get() is safe
            Set<AccessControlEntry> accessControlEntries = allAclsInSnapshot.get(resourcePattern);
            String logMsg = "Wrote " + accessControlEntries.size() + " for resource " + resourcePattern;
            operationConsumer.accept(UPDATE_ACL, logMsg, migrationState ->
                migrationClient.aclClient().writeResourceAcls(resourcePattern, accessControlEntries, migrationState));
        });

        resourcesToDelete.forEach(deletedResource -> {
            String logMsg = "Deleted resource " + deletedResource + " which has no ACLs in snapshot";
            operationConsumer.accept(DELETE_ACL, logMsg, migrationState ->
                migrationClient.aclClient().deleteResource(deletedResource, migrationState));
        });

        changedResources.forEach((resourcePattern, accessControlEntries) -> {
            String logMsg = "Wrote " + accessControlEntries.size() + " for resource " + resourcePattern;
            operationConsumer.accept(UPDATE_ACL, logMsg, migrationState ->
                migrationClient.aclClient().writeResourceAcls(resourcePattern, accessControlEntries, migrationState));
        });
    }
Handle a snapshot of the topic metadata. This requires scanning through all the topics and partitions
in ZooKeeper to determine what has changed. Topic configs are not handled here since they exist in the
ConfigurationsImage.
metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationZkWriter.java