def check_rpc_bipartite_etype_sampling_empty(
    tmpdir, num_server, use_graphbolt=False, return_eids=False
):
    """sample on bipartite via sample_etype_neighbors() which yields empty sample results"""
    generate_ip_config("rpc_ip_config.txt", num_server, num_server)

    g = create_random_bipartite()
    num_parts = num_server
    num_hops = 1

    orig_nids, _ = partition_graph(
        g,
        "test_sampling",
        num_parts,
        tmpdir,
        num_hops=num_hops,
        part_method="metis",
        return_mapping=True,
        use_graphbolt=use_graphbolt,
        store_eids=return_eids,
    )

    pserver_list = []
    ctx = mp.get_context("spawn")
    for i in range(num_server):
        p = ctx.Process(
            target=start_server,
            args=(
                i,
                tmpdir,
                num_server > 1,
                "test_sampling",
                ["csc", "coo"],
                use_graphbolt,
            ),
        )
        p.start()
        time.sleep(1)
        pserver_list.append(p)

    deg = get_degrees(g, orig_nids["game"], "game")
    empty_nids = F.nonzero_1d(deg == 0).to(g.idtype)
    nodes = {"game": empty_nids, "user": torch.tensor([1], dtype=g.idtype)}
    block, _ = start_bipartite_etype_sample_client(
        0,
        tmpdir,
        num_server > 1,
        nodes=nodes,
        use_graphbolt=use_graphbolt,
        return_eids=return_eids,
    )

    print("Done sampling")
    for p in pserver_list:
        p.join()
        assert p.exitcode == 0

    assert block is not None
    assert block.num_edges() == 0
    assert len(block.etypes) == len(g.etypes)
sample on bipartite via sample_etype_neighbors() which yields empty sample results
tests/distributed/test_distributed_sampling.py