{"record":{"id":"36d7f5126c6e18ec","repo":"mlflow/mlflow","slug":"asyncloggingqueue-is-not-activated","errorCode":null,"errorMessage":"AsyncLoggingQueue is not activated.","messagePattern":"AsyncLoggingQueue is not activated\\.","errorType":"exception","errorClass":"MlflowException","httpStatus":null,"severity":"error","filePath":"mlflow/utils/async_logging/async_logging_queue.py","lineNumber":305,"sourceCode":"        \"\"\"Asynchronously logs a batch of run data (parameters, tags, and metrics).\n\n        Args:\n            run_id (str): The ID of the run to log data for.\n            params (list[mlflow.entities.Param]): A list of parameters to log for the run.\n            tags (list[mlflow.entities.RunTag]): A list of tags to log for the run.\n            metrics (list[mlflow.entities.Metric]): A list of metrics to log for the run.\n\n        Returns:\n            mlflow.utils.async_utils.RunOperations: An object that encapsulates the\n                asynchronous operation of logging the batch of run data.\n                The object contains a list of `concurrent.futures.Future` objects that can be used\n                to check the status of the operation and retrieve any exceptions\n                that occurred during the operation.\n        \"\"\"\n        from mlflow import MlflowException\n\n        if not self.is_active():\n            raise MlflowException(\"AsyncLoggingQueue is not activated.\")\n        batch = RunBatch(\n            run_id=run_id,\n            params=params,\n            tags=tags,\n            metrics=metrics,\n            completion_event=threading.Event(),\n        )\n        self._queue.put(batch)\n        operation_future = self._batch_status_check_threadpool.submit(self._wait_for_batch, batch)\n        return RunOperations(operation_futures=[operation_future])\n\n    def is_active(self) -> bool:\n        return self._status == QueueStatus.ACTIVE\n\n    def is_idle(self) -> bool:\n        return self._status == QueueStatus.IDLE\n\n    def _set_up_logging_thread(self) -> None:","sourceCodeStart":287,"sourceCodeEnd":323,"githubUrl":"https://github.com/mlflow/mlflow/blob/6a27f2decc0b76eb1b54af31849784addb357dbc/mlflow/utils/async_logging/async_logging_queue.py#L287-L323","documentation":"AsyncLoggingQueue.log_batch_async checks `is_active()` before enqueueing a RunBatch. If the queue's consumer thread was never started (`start()` not called) or has been stopped/terminated, MLflow raises MlflowException because there is no worker to consume the batch. This fails fast to avoid silently losing logged metrics, params, or tags.","triggerScenarios":"Calling `log_batch_async` on an AsyncLoggingQueue (e.g. AsyncBatchLoggingQueue used by MlflowClient for async logging) before `start()`, or after `stop()`/termination, or on a queue object whose worker thread died.","commonSituations":"Manually constructing an AsyncLoggingQueue without starting it; calling async logging after client teardown; sharing a queue across forked/pickled contexts where the thread doesn't survive; enabling async logging (MLFLOW_ENABLE_ASYNC_LOGGING) but terminating the run/flushing too early.","solutions":["Call `queue.start()` before the first `log_batch_async` call.","Prefer the managed MlflowClient async path so activation is handled automatically.","If the queue was stopped, instantiate and start a fresh queue instead of reusing it.","Check `queue.is_active()` before enqueueing and fall back to synchronous `log_batch` when inactive."],"exampleFix":"# before\nqueue = AsyncLoggingQueue()\nqueue.log_batch_async(batch)\n\n# after\nqueue = AsyncLoggingQueue()\nqueue.start()\nqueue.log_batch_async(batch)","handlingStrategy":"validation","validationCode":"if not queue.is_active():\n    queue.start()\nqueue.log_batch_async(run_id, metrics=metrics, params=params, tags=tags)","typeGuard":"def is_logging_queue_ready(queue) -> bool:\n    return callable(getattr(queue, \"is_active\", None)) and queue.is_active()","tryCatchPattern":"from mlflow.exceptions import MlflowException\ntry:\n    queue.log_batch_async(run_id, metrics=metrics, params=params, tags=tags)\nexcept MlflowException as e:\n    if \"not activated\" in str(e):\n        queue.start()\n        queue.log_batch_async(run_id, metrics=metrics, params=params, tags=tags)\n    else:\n        raise","preventionTips":["Call start() once at client initialization before any async logging.","Check is_active() as a guard before every log_batch_async call.","Avoid forking or pickling queues with live consumer threads; recreate after fork.","Ensure stop()/flush is called on shutdown before process exit to avoid lost batches."],"tags":["async","lifecycle","queue","logging"],"backgroundTag":"queue-not-activated","analyzedSha":"6a27f2decc0b76eb1b54af31849784addb357dbc","analyzedAt":"2026-08-29T20:54:51.419Z","schemaVersion":2},"datasetVersion":"2026-08-29T22:17:34.462Z"}