code: def initialize_and_test_parameters(
      rank: int,
      world_size: int,
      backend: str,
      embedding_tables: Union[
          EmbeddingCollection, EmbeddingBagCollection
      ],
      sharding_type: str,
      sharders: List[ModuleSharder[nn.Module]],
      table_name: str,
      local_size: Optional[int] = None,
  ) -> None:
      with MultiProcessContext(
          rank, world_size, backend, local_size
      ) as ctx:
          module_sharding_plan = construct_module_sharding_plan(
              embedding_tables,
              per_param_sharding={
                  table_name: _select_sharding_type(sharding_type),
              },
              local_size=ctx.local_size,
              world_size=ctx.world_size,
              device_type=ctx.device.type,
          )
  
          model = DistributedModelParallel(
              module=embedding_tables,
              plan=ShardingPlan({"": module_sharding_plan}),
              env=ShardingEnv.from_process_group(ctx.pg),
              sharders=sharders,
              device=ctx.device,
          )
  
          key = (
              f"embeddings.{table_name}.weight"
              if isinstance(embedding_tables, EmbeddingCollection)
              else f"embedding_bags.{table_name}.weight"
          )
  
          if isinstance(model.state_dict()[key], ShardedTensor):
              if ctx.rank == 0:
                  gathered_tensor = torch.empty_like(
                      embedding_tables.state_dict()[key]
                  )
              else:
                  gathered_tensor = None
  
              model.state_dict()[key].gather(dst=0, out=gathered_tensor)
  
              if ctx.rank == 0:
                  torch.testing.assert_close(
                      gathered_tensor,
                      embedding_tables.state_dict()[key],
                  )
          elif isinstance(model.state_dict()[key], torch.Tensor):
              torch.testing.assert_close(
                  embedding_tables.state_dict()[key].cpu(),
                  model.state_dict()[key].cpu(),
              )
          else:
              raise AssertionError(
                  f"Model state dict contains unsupported type for key: {key}"
              )
  
imports: import unittest
  from typing import cast, List, Optional, Union
  import torch
  from hypothesis import given, settings, strategies, Verbosity
  from torch import nn
  from torchrec.distributed.embedding import EmbeddingCollectionSharder
  from torchrec.distributed.embeddingbag import EmbeddingBagCollectionSharder
  from torchrec.distributed.model_parallel import DistributedModelParallel
  from torchrec.distributed.sharding_plan import column_wise, construct_module_sharding_plan, data_parallel, ParameterShardingGenerator, row_wise, table_wise
  from torchrec.distributed.test_utils.multi_process import MultiProcessContext, MultiProcessTestBase
  from torchrec.distributed.types import ModuleSharder, ShardedTensor, ShardingEnv, ShardingPlan, ShardingType
  from torchrec.modules.embedding_configs import DataType, EmbeddingBagConfig, EmbeddingConfig
  from torchrec.modules.embedding_modules import EmbeddingBagCollection, EmbeddingCollection
  from torchrec.test_utils import skip_if_asan_class
label_desc: Use image node_id 1 for calling a global function with example usage: initialize_and_test_parameters(rank, world_size, backend, embedding_tables, sharding_type, sharders, table_name, local_size) without return types