WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-08-24 16:42:24 +02:00
Code Issues Packages Projects Releases Wiki Activity

Update kafka

Browse Source
This commit is contained in:
William Fu-Hinthorn
2025-03-18 17:30:21 -07:00
parent d6caa3b00a
commit 12c7ddf3a3
1 changed files with 9 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/scheduler-kafka/langgraph/scheduler/kafka/executor.py
+9 -6
View File
@@ -1,6 +1,7 @@
import asyncio
import binascii
import concurrent.futures
import weakref
from collections.abc import Sequence
from contextlib import (
AbstractAsyncContextManager,
@@ -216,10 +217,11 @@ class AsyncKafkaExecutor(AbstractAsyncContextManager):
checkpoint_null_version=checkpoint_null_version(saved.checkpoint),
):
# execute task, saving writes
put_writes = partial(self._put_writes, submit, msg["config"])
runner = PregelRunner(
submit=submit,
put_writes=partial(self._put_writes, submit, msg["config"]),
schedule_task=self._schedule_task,
submit=weakref.ref(submit),
put_writes=weakref.ref(put_writes),
schedule_task=weakref.WeakMethod(self._schedule_task),
)
async for _ in runner.atick([task], reraise=False):
pass
@@ -432,10 +434,11 @@ class KafkaExecutor(AbstractContextManager):
checkpoint_null_version=checkpoint_null_version(saved.checkpoint),
):
# execute task, saving writes
put_writes = partial(self._put_writes, submit, msg["config"])
runner = PregelRunner(
submit=submit,
put_writes=partial(self._put_writes, submit, msg["config"]),
schedule_task=self._schedule_task,
submit=weakref.ref(submit),
put_writes=weakref.ref(put_writes),
schedule_task=weakref.WeakMethod(self._schedule_task),
)
for _ in runner.tick([task], reraise=False):
pass