Fixes io loop event exeception by closing gremlin python client (#788)

.gitignore

@@ -61,3 +61,4 @@ interactive_engine/src/executor/store/src/db/proto/*
 !interactive_engine/src/executor/store/src/db/proto/mod.rs
 interactive_engine/src/data_load_tools/dependency-reduced-pom.xml
 interactive_engine/src/gaia-adaptor/dependency-reduced-pom.xml
+interactive_engine/src/executor/Cargo.lock

coordinator/gscoordinator/coordinator.py

@@ -859,7 +859,7 @@ def _close_interactive_instance(self, op: op_def_pb2.OpDef):
             proc = self._launcher.close_interactive_instance(object_id)
             # 60s is enough
             proc.wait(timeout=60)
-            gremlin_client.closed = True
+            gremlin_client.close()
         except Exception as e:
             raise RuntimeError(
                 f"Failed to close interactive instance {object_id}"

coordinator/gscoordinator/object_manager.py

@@ -54,6 +54,14 @@ def submit(self, message, bindings=None, request_options=None, query_gaia=False)
         else:
             return self.clients[0].submit(message, bindings, request_options)
 
+    def close(self):
+        for client in self.clients:
+            try:
+                client.close()
+            except Exception:
+                pass
+        self.closed = True
+
 
 class GremlinResultSet(object):
     def __init__(self, key, result_set):

coordinator/gscoordinator/utils.py

@@ -1504,7 +1504,9 @@ def check_gremlin_server_ready(endpoint):
         except Exception as e:
             error_message = str(e)
         else:
+            client.close()
             return True
         time.sleep(3)
         if time.time() - begin_time > INTERAVTIVE_INSTANCE_TIMEOUT_SECONDS:
+            client.close()
             raise TimeoutError(f"Gremlin check query failed: {error_message}")