{"record":{"id":"861dab32a53272db","repo":"apache/beam","slug":"qdrant-client-is-not-initialized","errorCode":null,"errorMessage":"Qdrant client is not initialized","messagePattern":"Qdrant client is not initialized","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/ml/rag/ingestion/qdrant.py","lineNumber":297,"sourceCode":"        check_compatibility=False,\n        **params.kwargs,\n    )\n\n  def teardown(self):\n    if self._client:\n      try:\n        self._client.close()\n      finally:\n        self._client = None\n\n  def finish_bundle(self):\n    self._flush()\n\n  def _flush(self):\n    if not self._batch:\n      return\n    if not self._client:\n      raise RuntimeError(\"Qdrant client is not initialized\")\n\n    max_retries = 3\n    attempt = 1\n    while True:\n      try:\n        self._client.upsert(\n            collection_name=self.config.collection_name,\n            points=self._batch,\n            timeout=self.config.timeout,\n            **self.config.kwargs,\n        )\n        break\n      except ResourceExhaustedResponse as e:\n        time.sleep(e.retry_after_s)\n        # don't count rate-limit against max_retries\n        continue\n      except (UnexpectedResponse, ResponseHandlingException,\n              grpc.RpcError) as e:","sourceCodeStart":279,"sourceCodeEnd":315,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/ml/rag/ingestion/qdrant.py#L279-L315","documentation":"_flush() performs the actual Qdrant upsert using self._client. The client is created lazily in setUp/process (inside the DoFn runtime); if _flush is reached while _client is still None, the DoFn's internal invariant is broken and it raises RuntimeError instead of crashing the client call.","triggerScenarios":"Calling _flush() (via process or finish_bundle) on a _QdrantWriteTransform DoFn instance that never ran setUp / setup() — e.g. direct unit-test invocation of process/finish_bundle without setup, or a subclass overriding setup without initializing _client.","commonSituations":"Testing the DoFn directly without invoking setup(); runners or wrappers that skip DoFn lifecycle methods; custom subclasses of the transform that bypass client initialization.","solutions":["In tests, call dofn.setup() before processing elements (or use the transform via a proper pipeline/test pipeline)","Ensure any subclass override of setup() calls super().setup() so _client is initialized","Use a TestPipeline / beam.test pipeline so the runner runs the full DoFn lifecycle","If client creation depends on config, verify the config is valid so setup() doesn't silently skip client creation"],"exampleFix":"// before\ndofn = _QdrantWriteFn(config)\ndofn.process(item)\n// after\ndofn = _QdrantWriteFn(config)\ndofn.setup()\ndofn.process(item)","handlingStrategy":"try-catch","validationCode":"# in tests, before calling process/finish_bundle:\ndofn.setup()\nassert dofn._client is not None","typeGuard":"def is_ready(dofn) -> bool:\n    return getattr(dofn, \"_client\", None) is not None","tryCatchPattern":"try:\n    dofn.finish_bundle()\nexcept RuntimeError as e:\n    if \"not initialized\" in str(e):\n        dofn.setup()\n        dofn.finish_bundle()\n    else:\n        raise","preventionTips":["Exercise DoFns through a TestPipeline so setup()/teardown lifecycle runs","Call setup() explicitly in unit tests before process()","When subclassing, always call super().setup()"],"tags":["python","qdrant","beam","lifecycle"],"backgroundTag":"internal-invariant-violation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}