void handleClientQuotasSnapshot(ClientQuotasImage clientQuotasImage, ScramImage scramImage, KRaftMigrationOperationConsumer opConsumer) {
        Set<ClientQuotaEntity> changedNonUserEntities = new HashSet<>();
        Set<String> changedUsers = new HashSet<>();

        if (clientQuotasImage != null) {
            for (Entry<ClientQuotaEntity, ClientQuotaImage> entry : clientQuotasImage.entities().entrySet()) {
                ClientQuotaEntity entity = entry.getKey();
                if (entity.entries().containsKey(ClientQuotaEntity.USER) &&
                    !entity.entries().containsKey(ClientQuotaEntity.CLIENT_ID)) {
                    // Track regular user entities separately
                    // There should only be 1 entry in the list of type ClientQuotaEntity.USER
                    changedUsers.add(entity.entries().get(ClientQuotaEntity.USER));
                } else {
                    changedNonUserEntities.add(entity);
                }
            }
        }
        if (scramImage != null) {
            for (Entry<ScramMechanism, Map<String, ScramCredentialData>> mechanismEntry : scramImage.mechanisms().entrySet()) {
                for (Entry<String, ScramCredentialData> userEntry : mechanismEntry.getValue().entrySet()) {
                    changedUsers.add(userEntry.getKey());
                }
            }
        }
        migrationClient.configClient().iterateClientQuotas(new ConfigMigrationClient.ClientQuotaVisitor() {
            @Override
            public void visitClientQuota(List<ClientQuotaRecord.EntityData> entityDataList, Map<String, Double> quotas) {
                Map<String, String> entityMap = new HashMap<>(2);
                entityDataList.forEach(entityData -> entityMap.put(entityData.entityType(), entityData.entityName()));
                ClientQuotaEntity entity = new ClientQuotaEntity(entityMap);
                if (!clientQuotasImage.entities().getOrDefault(entity, ClientQuotaImage.EMPTY).quotaMap().equals(quotas)) {
                    if (entity.entries().containsKey(ClientQuotaEntity.USER) &&
                        !entity.entries().containsKey(ClientQuotaEntity.CLIENT_ID)) {
                        // Track regular user entities separately
                        changedUsers.add(entityMap.get(ClientQuotaEntity.USER));
                    } else {
                        changedNonUserEntities.add(entity);
                    }
                }
            }

            @Override
            public void visitScramCredential(String userName, ScramMechanism scramMechanism, ScramCredential scramCredential) {
                // For each ZK entity, see if it exists in the image and if it's equal
                ScramCredentialData data = scramImage.mechanisms().getOrDefault(scramMechanism, Collections.emptyMap()).get(userName);
                if (data == null || !data.toCredential(scramMechanism).equals(scramCredential)) {
                    changedUsers.add(userName);
                }
            }
        });

        changedNonUserEntities.forEach(entity -> {
            Map<String, Double> quotaMap = clientQuotasImage.entities().getOrDefault(entity, ClientQuotaImage.EMPTY).quotaMap();
            opConsumer.accept(UPDATE_CLIENT_QUOTA, "Updated client quotas for " + entity, migrationState ->
                migrationClient.configClient().writeClientQuotas(entity.entries(), quotaMap, Collections.emptyMap(), migrationState));
        });

        changedUsers.forEach(userName -> {
            ClientQuotaEntity entity = new ClientQuotaEntity(Collections.singletonMap(ClientQuotaEntity.USER, userName));
            Map<String, Double> quotaMap = clientQuotasImage.entities().
                getOrDefault(entity, ClientQuotaImage.EMPTY).quotaMap();
            Map<String, String> scramMap = getScramCredentialStringsForUser(scramImage, userName);
            opConsumer.accept(UPDATE_CLIENT_QUOTA, "Updated client quotas for " + userName, migrationState ->
                migrationClient.configClient().writeClientQuotas(entity.entries(), quotaMap, scramMap, 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