mirror of
https://git.mirrors.martin98.com/https://github.com/infiniflow/ragflow.git
synced 2025-07-15 23:51:46 +08:00
Fix get_unacked_iterator (#6280)
### What problem does this PR solve? Fix get_unacked_iterator. Close #6132 ### Type of change - [x] Bug Fix (non-breaking change which fixes an issue)
This commit is contained in:
parent
9cad60fa6d
commit
bb869aca33
@ -592,6 +592,7 @@ async def handle_task():
|
|||||||
global DONE_TASKS, FAILED_TASKS
|
global DONE_TASKS, FAILED_TASKS
|
||||||
redis_msg, task = await collect()
|
redis_msg, task = await collect()
|
||||||
if not task:
|
if not task:
|
||||||
|
await trio.sleep(5)
|
||||||
return
|
return
|
||||||
try:
|
try:
|
||||||
logging.info(f"handle_task begin for task {json.dumps(task)}")
|
logging.info(f"handle_task begin for task {json.dumps(task)}")
|
||||||
|
@ -209,7 +209,7 @@ class RedisDB:
|
|||||||
"""https://redis.io/docs/latest/commands/xreadgroup/"""
|
"""https://redis.io/docs/latest/commands/xreadgroup/"""
|
||||||
try:
|
try:
|
||||||
group_info = self.REDIS.xinfo_groups(queue_name)
|
group_info = self.REDIS.xinfo_groups(queue_name)
|
||||||
if not any(e["name"] == group_name for e in group_info):
|
if not any(gi["name"] == group_name for gi in group_info):
|
||||||
self.REDIS.xgroup_create(queue_name, group_name, id="0", mkstream=True)
|
self.REDIS.xgroup_create(queue_name, group_name, id="0", mkstream=True)
|
||||||
args = {
|
args = {
|
||||||
"groupname": group_name,
|
"groupname": group_name,
|
||||||
@ -228,7 +228,7 @@ class RedisDB:
|
|||||||
res = RedisMsg(self.REDIS, queue_name, group_name, msg_id, payload)
|
res = RedisMsg(self.REDIS, queue_name, group_name, msg_id, payload)
|
||||||
return res
|
return res
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
if "key" in str(e):
|
if str(e) == 'no such key':
|
||||||
pass
|
pass
|
||||||
else:
|
else:
|
||||||
logging.exception(
|
logging.exception(
|
||||||
@ -242,8 +242,14 @@ class RedisDB:
|
|||||||
def get_unacked_iterator(self, queue_names: list[str], group_name, consumer_name):
|
def get_unacked_iterator(self, queue_names: list[str], group_name, consumer_name):
|
||||||
try:
|
try:
|
||||||
for queue_name in queue_names:
|
for queue_name in queue_names:
|
||||||
group_info = self.REDIS.xinfo_groups(queue_name)
|
try:
|
||||||
if not any(e["name"] == group_name for e in group_info):
|
group_info = self.REDIS.xinfo_groups(queue_name)
|
||||||
|
except Exception as e:
|
||||||
|
if str(e) == 'no such key':
|
||||||
|
logging.warning(f"RedisDB.get_unacked_iterator queue {queue_name} doesn't exist")
|
||||||
|
continue
|
||||||
|
if not any(gi["name"] == group_name for gi in group_info):
|
||||||
|
logging.warning(f"RedisDB.get_unacked_iterator queue {queue_name} group {group_name} doesn't exist")
|
||||||
continue
|
continue
|
||||||
current_min = 0
|
current_min = 0
|
||||||
while True:
|
while True:
|
||||||
@ -251,13 +257,11 @@ class RedisDB:
|
|||||||
if not payload:
|
if not payload:
|
||||||
break
|
break
|
||||||
current_min = payload.get_msg_id()
|
current_min = payload.get_msg_id()
|
||||||
logging.info(f"RedisDB.get_unacked_iterator {consumer_name} msg_id {current_min}")
|
logging.info(f"RedisDB.get_unacked_iterator {queue_name} {consumer_name} {current_min}")
|
||||||
yield payload
|
yield payload
|
||||||
except Exception as e:
|
except Exception:
|
||||||
if "key" in str(e):
|
|
||||||
return
|
|
||||||
logging.exception(
|
logging.exception(
|
||||||
"RedisDB.get_unacked_iterator " + consumer_name + " got exception: "
|
"RedisDB.get_unacked_iterator got exception: "
|
||||||
)
|
)
|
||||||
self.__open__()
|
self.__open__()
|
||||||
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user