void handleConfigsSnapshot(ConfigurationsImage configsImage, KRaftMigrationOperationConsumer operationConsumer) {
        Set<ConfigResource> newResources = new HashSet<>();
        configsImage.resourceData().keySet().forEach(resource -> {
            if (EnumSet.of(ConfigResource.Type.BROKER, ConfigResource.Type.TOPIC).contains(resource.type())) {
                newResources.add(resource);
            } else {
                throw new RuntimeException("Unknown config resource type " + resource.type());
            }
        });
        Set<ConfigResource> resourcesToUpdate = new HashSet<>();
        BiConsumer<ConfigResource, Map<String, String>> processConfigsForResource = (ConfigResource resource, Map<String, String> configs) -> {
            newResources.remove(resource);
            Map<String, String> kraftProps = configsImage.configMapForResource(resource);
            if (!kraftProps.equals(configs)) {
                resourcesToUpdate.add(resource);
            }
        };

        migrationClient.configClient().iterateBrokerConfigs((broker, configs) -> {
            ConfigResource brokerResource = new ConfigResource(ConfigResource.Type.BROKER, broker);
            processConfigsForResource.accept(brokerResource, configs);
        });
        migrationClient.configClient().iterateTopicConfigs((topic, configs) -> {
            ConfigResource topicResource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
            processConfigsForResource.accept(topicResource, configs);
        });

        newResources.forEach(resource -> {
            Map<String, String> props = configsImage.configMapForResource(resource);
            if (!props.isEmpty()) {
                String opType = brokerOrTopicOpType(resource, UPDATE_BROKER_CONFIG, UPDATE_TOPIC_CONFIG);
                operationConsumer.accept(opType, "Created configs for " + resource.type().name() + " " + resource.name(),
                    migrationState -> migrationClient.configClient().writeConfigs(resource, props, migrationState));
            }
        });

        resourcesToUpdate.forEach(resource -> {
            Map<String, String> props = configsImage.configMapForResource(resource);
            if (props.isEmpty()) {
                String opType = brokerOrTopicOpType(resource, DELETE_BROKER_CONFIG, DELETE_TOPIC_CONFIG);
                operationConsumer.accept(opType, "Deleted configs for " + resource.type().name() + " " + resource.name(),
                    migrationState -> migrationClient.configClient().deleteConfigs(resource, migrationState));
            } else {
                String opType = brokerOrTopicOpType(resource, UPDATE_BROKER_CONFIG, UPDATE_TOPIC_CONFIG);
                operationConsumer.accept(opType, "Updated configs for " + resource.type().name() + " " + resource.name(),
                    migrationState -> migrationClient.configClient().writeConfigs(resource, props, 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