WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-13 21:27:52 +02:00
Code Issues Packages Projects Releases Wiki Activity

scheduler-kafka: Pass loop arg to default async consumer and producer

Browse Source 
- Some forks of aiokafka make this a required arg
This commit is contained in:
Nuno Campos
2024-10-07 10:16:36 -07:00
parent debfd85ff8
commit e5b4cd2701
2 changed files with 6 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/scheduler-kafka/langgraph/scheduler/kafka/executor.py
+3
View File
@@ -69,6 +69,7 @@ class AsyncKafkaExecutor(AbstractAsyncContextManager):
self.retry_policy = retry_policy
async def __aenter__(self) -> Self:
loop = asyncio.get_running_loop()
self.subgraphs = {
k: v async for k, v in self.graph.aget_subgraphs(recurse=True)
}
@@ -81,6 +82,7 @@ class AsyncKafkaExecutor(AbstractAsyncContextManager):
auto_offset_reset="earliest",
group_id="executor",
enable_auto_commit=False,
loop=loop,
**self.kwargs,
)
)
@@ -89,6 +91,7 @@ class AsyncKafkaExecutor(AbstractAsyncContextManager):
self.producer = await self.stack.enter_async_context(
DefaultAsyncProducer(
loop=loop,
**self.kwargs,
)
)
libs/scheduler-kafka/langgraph/scheduler/kafka/orchestrator.py
+3
View File
@@ -67,6 +67,7 @@ class AsyncKafkaOrchestrator(AbstractAsyncContextManager):
self.retry_policy = retry_policy
async def __aenter__(self) -> Self:
loop = asyncio.get_running_loop()
self.subgraphs = {
k: v async for k, v in self.graph.aget_subgraphs(recurse=True)
}
@@ -79,6 +80,7 @@ class AsyncKafkaOrchestrator(AbstractAsyncContextManager):
auto_offset_reset="earliest",
group_id="orchestrator",
enable_auto_commit=False,
loop=loop,
**self.kwargs,
)
)
@@ -87,6 +89,7 @@ class AsyncKafkaOrchestrator(AbstractAsyncContextManager):
self.producer = await self.stack.enter_async_context(
DefaultAsyncProducer(
loop=loop,
**self.kwargs,
)
)