What happened?
In the Milvus RAG vector sink (apache_beam/ml/rag/ingestion/milvus_search.py), _MilvusSink.write() creates a new MilvusClient on every call:
def write(self, documents):
self._client = MilvusClient(
**unpack_dataclass_with_kwargs(self._connection_params))
resp = self._client.upsert(...)
But __enter__ already creates the client using retry_with_backoff. So the write path:
- throws away the retry-wrapped client from
__enter__ and makes a fresh one without retry
- leaves the original client unclosed, so it leaks (MilvusClient has no
__del__)
__exit__ only closes the last client
The write() docstring also says it "flushes the collection to ensure data persistence", but there is no flush() call anywhere.
Expected: write() should reuse the client from __enter__, and the docstring should describe the real behavior (no flush).
Issue Priority
Priority: 3 (minor)
Issue Components
What happened?
In the Milvus RAG vector sink (
apache_beam/ml/rag/ingestion/milvus_search.py),_MilvusSink.write()creates a newMilvusClienton every call:But
__enter__already creates the client usingretry_with_backoff. So the write path:__enter__and makes a fresh one without retry__del__)__exit__only closes the last clientThe
write()docstring also says it "flushes the collection to ensure data persistence", but there is noflush()call anywhere.Expected:
write()should reuse the client from__enter__, and the docstring should describe the real behavior (no flush).Issue Priority
Priority: 3 (minor)
Issue Components