jina-ai · hanxiao · Feb 4, 2021 · Feb 4, 2021
diff --git a/jina/peapods/pods/__init__.py b/jina/peapods/pods/__init__.py
@@ -196,10 +196,12 @@ def start(self) -> 'BasePod':
         which is inherited from :class:`jina.peapods.peas.BasePea`
         """
         # start head and tail
-
-        for _args in self.all_args:
-            self._enter_pea(BasePea(_args))
-
+        try:
+            for _args in self.all_args:
+                self._enter_pea(BasePea(_args))
+        except:
+            self.close()
+            raise
         return self
 
     def _enter_pea(self, pea: 'BasePea') -> None:

diff --git a/tests/daemon/unit/stores/test_flowstore.py b/tests/daemon/unit/stores/test_flowstore.py
@@ -1,7 +1,5 @@
 from pathlib import Path
 
-from fastapi import UploadFile
-
 from daemon.stores import FlowStore
 from jina.flow import Flow
 
@@ -22,7 +20,7 @@ def test_flow_store():
     assert len(store) == 1
     assert flow_id in store
     assert isinstance(store[flow_id]['object'], Flow)
-    del store[flow_id]
+    store.delete(flow_id)
     assert flow_id not in store
     assert not store
 
diff --git a/tests/integration/issues/github_1861/__init__.py b/tests/integration/issues/github_1861/__init__.py
diff --git a/tests/integration/issues/github_1861/test_pea_closing.py b/tests/integration/issues/github_1861/test_pea_closing.py
@@ -0,0 +1,22 @@
+import pytest
+
+from jina.executors.encoders import BaseEncoder
+from jina.flow import Flow
+
+
+class ExceptionExecutor(BaseEncoder):
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.name = 'exception-executor'
+
+    def post_init(self):
+        raise Exception
+
+
+@pytest.mark.timeout(10)
+def test_pea_closing():
+    with pytest.raises(Exception):
+        with Flow().add(uses='!ExceptionExecutor', parallel=3) as f:
+            pod1 = f._pod_nodes['pod0']
+            assert len(pod1.peas) == 0