Repository: dask/dask
File: dask/array/core.py (Python)

Code before:
        zarr_mem_store_types = (dict, zarr.storage.MemoryStore, zarr.storage.KVStore)

    if isinstance(url, zarr.Array):
        z = url
        if isinstance(z.store, zarr_mem_store_types):
            try:
                from distributed import default_client

                default_client()
            except (ImportError, ValueError):
                pass
            else:
                raise RuntimeError(
                    "Cannot store into in memory Zarr Array using "
                    "the distributed scheduler."
                )
        zarr_write_chunks = _get_zarr_write_chunks(z)
        dask_write_chunks = normalize_chunks(
            chunks="auto",
            shape=z.shape,
            dtype=z.dtype,
            previous_chunks=zarr_write_chunks,
        )

        for ax, (dw, zw) in enumerate(
            zip(dask_write_chunks, zarr_write_chunks, strict=False)
        ):
            if len(dw) >= 1:
                nominal_dask_chunk_size = dw[0]
                if not nominal_dask_chunk_size % zw == 0:
                    safe_chunk_size = np.prod(zarr_write_chunks) * max(
                        1, z.dtype.itemsize
                    )
                    msg = (
                        f"The input Dask array will be rechunked along axis {ax} with chunk size "
                        f"{nominal_dask_chunk_size}, but a chunk size divisible by {zw} is "
                        f"required for Dask to write safely to the Zarr array {z}. "
                        "To avoid risk of data loss when writing to this Zarr array, set the "
                        '"array.chunk-size" configuration parameter to at least the size in'
                        " bytes of a single on-disk "
                        f"chunk of the Zarr array, which in this case is "
                        f"{safe_chunk_size} bytes. "
                        f'E.g., dask.config.set({{"array.chunk-size": {safe_chunk_size}}})'
                    )
                    raise PerformanceWarning(msg)
                    break
        if region is None:
            # Get the appropriate write granularity (shard shape if sharding, else chunk shape)
            arr = arr.rechunk(dask_write_chunks)
            regions = None
        else:

Review comment (suggestion):
```suggestion
            zip(dask_write_chunks, zarr_write_chunks, strict=True)
```

Code after:
    """
    import zarr

    if np.isnan(arr.shape).any():
        raise ValueError(
            "Saving a dask array with unknown chunk sizes is not "
            "currently supported by Zarr.%s" % unknown_chunk_message
        )

    if _zarr_v3():
        zarr_mem_store_types = (zarr.storage.MemoryStore,)
    else:
        zarr_mem_store_types = (dict, zarr.storage.MemoryStore, zarr.storage.KVStore)

    if isinstance(url, zarr.Array):
        z = url
        if isinstance(z.store, zarr_mem_store_types):
            try:
                from distributed import default_client

                default_client()
            except (ImportError, ValueError):
                pass
            else:
                raise RuntimeError(
                    "Cannot store into in memory Zarr Array using "
                    "the distributed scheduler."
                )
        zarr_write_chunks = _get_zarr_write_chunks(z)
        dask_write_chunks = normalize_chunks(
            chunks="auto",
            shape=z.shape,
            dtype=z.dtype,
            previous_chunks=zarr_write_chunks,
        )

        for ax, (dw, zw) in enumerate(
            zip(dask_write_chunks, zarr_write_chunks, strict=True)
        ):
            if len(dw) >= 1:
                nominal_dask_chunk_size = dw[0]
                if not nominal_dask_chunk_size % zw == 0:
                    safe_chunk_size = np.prod(zarr_write_chunks) * max(
                        1, z.dtype.itemsize
                    )
                    msg = (
                        f"The input Dask array will be rechunked along axis {ax} with chunk size "
                        f"{nominal_dask_chunk_size}, but a chunk size divisible by {zw} is "
                        f"required for Dask to write safely to the Zarr array {z}. "
                        "To avoid risk of data loss when writing to this Zarr array, set the "
                        '"array.chunk-size" configuration parameter to at least the size in'