async def METHOD_NAME(ray_start_regular_shared, create_cluster):
    from xgboost_ray import RayDMatrix, RayParams, train, predict
    from sklearn.datasets import load_breast_cancer
    assert create_cluster.session
    session = new_session(address=create_cluster.address, backend="ray")
    with session:
        train_x, train_y = load_breast_cancer(return_X_y=True, as_frame=True)
        df: md.DataFrame = md.concat(
            [md.DataFrame(train_x), md.DataFrame(train_y)], axis=1
        )
        df.execute()
        num_shards = 4
        ds = mdd.to_ray_mldataset(df, num_shards)
        assert isinstance(ds, ml_dataset.MLDataset)
        import gc
        gc.collect()  # Ensure MLDataset does hold mars dataframe to avoid gc.
        # train
        train_set = RayDMatrix(ds, "target")
        evals_result = {}
        bst = train(
            {
                "objective": "binary:logistic",
                "eval_metric": ["logloss", "error"],
            },
            train_set,
            evals_result=evals_result,
            evals=[(train_set, "train")],
            verbose_eval=False,
            ray_params=RayParams(
                num_actors=num_shards, cpus_per_actor=1  # Number of remote actors
            ),
        )
        bst.save_model("model.xgb")
        assert os.path.exists("model.xgb")
        os.remove("model.xgb")
        print("Final training error: {:.4f}".format(evals_result["train"]["error"][-1]))
        predict(bst, train_set, ray_params=RayParams(num_actors=2))