Skip bad pending sends

This commit is contained in:
Nuno Campos
2024-06-28 08:59:26 -07:00
parent 2bf4a72d9d
commit c326ad019d
@@ -1816,6 +1816,9 @@ def _prepare_next_tasks(
tasks: Union[list[PregelTaskDescription], list[PregelExecutableTask]] = []
# Consume pending packets
for packet in checkpoint["pending_sends"]:
if not isinstance(packet, Send):
logger.warn(f"Ignoring invalid packet type {type(packet)} in pending sends")
continue
if for_execution:
if node := processes[packet.node].get_node():
writes = deque()