code: def test_sharding_ec_per_parameter_optimizer(
      self,
      sharding_type: str,
  ) -> None:
      WORLD_SIZE = 2
  
      embedding_config = [
          EmbeddingConfig(
              name="table_0",
              feature_names=["feature_0"],
              embedding_dim=4,
              num_embeddings=4,
          ),
          EmbeddingConfig(
              name="table_1",
              feature_names=["feature_1"],
              embedding_dim=4,
              num_embeddings=4,
          ),
      ]
  
      # Rank 0
      #             instance 0   instance 1  instance 2
      # "feature_0"   [0, 1]       None        [2]
      # "feature_1"   [0, 1]       None        [2]
  
      # Rank 1
  
      #             instance 0   instance 1  instance 2
      # "feature_0"   [3, 2]       [1,2]        [0, 1,2,3]
      # "feature_1"   [2,3]       None        [2]
  
      kjt_input_per_rank = [  # noqa
          KeyedJaggedTensor.from_lengths_sync(
              keys=["feature_0", "feature_1"],
              values=torch.LongTensor([0, 1, 2, 0, 1, 2]),
              lengths=torch.LongTensor([2, 0, 1, 2, 0, 1]),
          ),
          KeyedJaggedTensor.from_lengths_sync(
              keys=["feature_0", "feature_1"],
              values=torch.LongTensor(
                  [3, 2, 1, 2, 0, 1, 2, 3, 2, 3, 2]
              ),
              lengths=torch.LongTensor([2, 2, 4, 2, 0, 1]),
          ),
      ]
  
      self._run_multi_process_test(
          callable=_test_sharding_ec,
          world_size=WORLD_SIZE,
          tables=embedding_config,
          initial_state_dict={
              "embeddings.table_0.weight": torch.Tensor(
                  [
                      [1, 1, 1, 1],
                      [2, 2, 2, 2],
                      [4, 4, 4, 4],
                      [8, 8, 8, 8],
                  ]
              ),
              "embeddings.table_1.weight": torch.Tensor(
                  [
                      [101, 101, 101, 101],
                      [102, 102, 102, 102],
                      [104, 104, 104, 104],
                      [108, 108, 108, 108],
                  ]
              ),
          },
          kjt_input_per_rank=kjt_input_per_rank,
          sharder=TestEmbeddingCollectionSharder(
              sharding_type=sharding_type
          ),
          backend="nccl",
      )
  
imports: import copy
  import unittest
  from typing import Any, Dict, List, Optional
  import hypothesis.strategies
  import torch
  import torch.nn
  from hypothesis import given, settings, Verbosity
  from torchrec import distributed, EmbeddingConfig
  from torchrec.distributed import DistributedModelParallel
  from torchrec.distributed.embedding import EmbeddingCollectionSharder
  from torchrec.distributed.embedding_lookup import EmbeddingComputeKernel
  from torchrec.distributed.embeddingbag import EmbeddingBagCollectionSharder
  from torchrec.distributed.planner import EmbeddingShardingPlanner, ParameterConstraints, Topology
  from torchrec.distributed.test_utils.multi_process import MultiProcessContext, MultiProcessTestBase
  from torchrec.distributed.test_utils.test_sharding import copy_state_dict
  from torchrec.distributed.types import ModuleSharder, QuantizedCommCodecs, ShardingEnv, ShardingPlan, ShardingType
  from torchrec.modules.embedding_configs import EmbeddingBagConfig
  from torchrec.modules.embedding_modules import EmbeddingBagCollection, EmbeddingCollection
  from torchrec.optim.apply_optimizer_in_backward import apply_optimizer_in_backward
  from torchrec.sparse.jagged_tensor import KeyedJaggedTensor
  from torchrec.test_utils import skip_if_asan_class
label_desc: Use image node_id 1 for calling the ShardedEmbeddingCollectionApplyOptimPerParamTest obj's underlying member method code with example usage: obj.test_sharding_ec_per_parameter_optimizer(sharding_type) without return types