code: def test_rank_assignment() -> None:
      from distributed import Client, LocalCluster
  
      def local_test(worker_id):
          with xgb.dask.CommunicatorContext(**args) as ctx:
              task_id = ctx["DMLC_TASK_ID"]
              matched = re.search(".*-([0-9]).*", task_id)
              rank = xgb.collective.get_rank()
              # As long as the number of workers is lesser than 10, rank and worker id
              # should be the same
              assert rank == int(matched.group(1))
  
      with LocalCluster(n_workers=8) as cluster:
          with Client(cluster) as client:
              workers = tm.get_client_workers(client)
              args = client.sync(
                  xgb.dask._get_rabit_args,
                  len(workers),
                  None,
                  client,
              )
  
              futures = client.map(
                  local_test, range(len(workers)), workers=workers
              )
              client.gather(futures)
  
file_path: xgboost/tests/python/test_tracker.py
imports: import re
  import sys
  import numpy
  import pytest
  import xgboost
  from xgboost import RabitTracker, collective
  from xgboost import testing
label_desc: Use image node_id 8 for calling a global function with example usage: test_rank_assignment() without return types