fix(pods): remove random identity assignment (#1755)

* fix(pods): remove random identity assignment

jina/peapods/pods/helper.py

@@ -17,7 +17,6 @@ def _set_peas_args(args: Namespace, head_args: Namespace = None, tail_args: Name
         if tail_args:
             _args.port_out = tail_args.port_in
         _args.port_ctrl = random_port()
-        _args.identity = random_identity()
         _args.socket_out = SocketType.PUSH_CONNECT
         if args.polling.is_push:
             if args.scheduling == SchedulerType.ROUND_ROBIN:
@@ -63,7 +62,6 @@ def _copy_to_head_args(args: Namespace, is_push: bool, as_router: bool = True) -
     _head_args = copy.deepcopy(args)
     _head_args.port_ctrl = random_port()
     _head_args.port_out = random_port()
-    _head_args.identity = random_identity()
     _head_args.uses = None
     if is_push:
         if args.scheduling == SchedulerType.ROUND_ROBIN:
@@ -100,7 +98,6 @@ def _copy_to_tail_args(args: Namespace, as_router: bool = True) -> Namespace:
     _tail_args = copy.deepcopy(args)
     _tail_args.port_in = random_port()
     _tail_args.port_ctrl = random_port()
-    _tail_args.identity = random_identity()
     _tail_args.socket_in = SocketType.PULL_BIND
     _tail_args.uses = None
 

tests/unit/flow/test_flow.py

@@ -554,3 +554,15 @@ def test_flow_workspace_id():
     f.workspace_id = new_id
     assert len(set(f.workspace_id.values())) == 1
     assert list(f.workspace_id.values())[0] == new_id
+
+
+def test_flow_identity_override():
+    f = Flow().add().add(parallel=2).add(parallel=2)
+
+    with f:
+        assert len(set(p.args.identity for _, p in f)) == f.num_pods
+
+    f = Flow(identity='123456').add().add(parallel=2).add(parallel=2)
+
+    with f:
+        assert len(set(p.args.identity for _, p in f)) == 1