async def METHOD_NAME(c, s, a, b):
    retries = 5
    plugin = ExampleAnnotationPlugin(retries=retries)
    s.add_plugin(plugin)
    assert plugin in s.plugins.values()
    df = dd.from_pandas(
        pd.DataFrame(
            {"a": np.arange(10, dtype=int), "b": np.arange(10, 0, -1, dtype=float)}
        ),
        npartitions=5,
    )
    df = df.shuffle("a", shuffle="tasks", max_branch=2)
    acol = df["a"]
    bcol = df["b"]
    with dask.annotate(retries=retries):
        df = acol + bcol
    with dask.config.set(optimization__fuse__active=False):
        rdf = await c.compute(df)
    assert rdf.dtypes == np.float64
    assert (rdf == 10.0).all()
    # There is an annotation match per partition (i.e. task)
    assert plugin.retry_matches == df.npartitions