From 8b7069f10e1b74a531b0edd976db8d295c3ce06e Mon Sep 17 00:00:00 2001 From: Wen Guan Date: Tue, 23 Jan 2024 11:08:08 +0100 Subject: [PATCH] fix conductor by marking missing messages as processed --- main/lib/idds/agents/conductor/conductor.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/main/lib/idds/agents/conductor/conductor.py b/main/lib/idds/agents/conductor/conductor.py index 226306a0..8e0ceab8 100644 --- a/main/lib/idds/agents/conductor/conductor.py +++ b/main/lib/idds/agents/conductor/conductor.py @@ -190,6 +190,8 @@ def get_output_messages(self): def is_message_processed(self, message): retries = message['retries'] try: + if message['status'] in [MessageStatus.New]: + return False if retries >= self.max_retries: self.logger.info("message %s has reached max retries %s" % (message['msg_id'], self.max_retries)) return True @@ -237,7 +239,7 @@ def is_message_processed(self, message): content_statuses = proc_conents.get(map_id, []) if not content_statuses: pass - if len(content_statuses) == 1 and content_statuses == [ContentStatus.New]: + if (len(content_statuses) == 1 and content_statuses == [ContentStatus.New]) or ContentStatus.Missing in content_statuses: all_map_id_processed = False return all_map_id_processed return all_map_id_processed