cocoindex-io/cocoindex · error · ImportError

confluent_kafka is required to use the Kafka connector. Plea

Error message

confluent_kafka is required to use the Kafka connector. Please install cocoindex[kafka].

What it means

Importing the Kafka source connector module fails because the optional third-party package confluent_kafka is not installed. CocoIndex keeps the Kafka connector optional so the core install stays light; importing the module without the dependency raises ImportError with an actionable message pointing to the 'kafka' extra.

Source

Thrown at python/cocoindex/connectors/kafka/_source.py:24

primitive stream (with a ``payloads()`` view yielding bytes), and
``topic_as_map`` interprets messages as a keyed map for use with ``mount_each``.

User-facing docs and worked examples:
https://cocoindex.io/docs/connectors/kafka
"""

from __future__ import annotations

import asyncio
import logging
from collections import deque
from typing import Callable

try:
    from confluent_kafka import Message, TopicPartition  # type: ignore[import-not-found]
    from confluent_kafka.aio import AIOConsumer  # type: ignore[import-not-found]
except ImportError as e:
    raise ImportError(
        "confluent_kafka is required to use the Kafka connector. "
        "Please install cocoindex[kafka]."
    ) from e

from cocoindex._internal.live_component import (
    _IMMEDIATE_READY,
    LiveMapSubscriber,
    LiveStream,
    LiveStreamSubscriber,
    ReadyAwaitable,
)
from cocoindex._internal.typing import StableKey
from cocoindex.connectorkits import SingleWatcherGuard

_logger = logging.getLogger(__name__)


# --- Public type aliases ---

View on GitHub (pinned to e84aa99b32)

Solutions

Install the extra: `pip install 'cocoindex[kafka]'` (or `uv add 'cocoindex[kafka]'`).
If already installed, verify `python -c "import confluent_kafka"` works; reinstall or fix librdkafka if it errors.
Add the extra to your project dependency file (pyproject.toml) so environments are reproducible.

Example fix

// before
from cocoindex.connectors.kafka import KafkaSource  # ImportError at import time
// after
# pip install 'cocoindex[kafka]'
from cocoindex.connectors.kafka import KafkaSource

Defensive patterns

Strategy: try-catch

Validate before calling

try:
    import confluent_kafka  # noqa: F401
    from confluent_kafka.aio import AIOConsumer  # noqa: F401
except ImportError:
    raise SystemExit("Install the Kafka extra: pip install 'cocoindex[kafka]'")

Try / catch

try:
    from cocoindex.connectors.kafka import KafkaSource
except ImportError as e:
    if "confluent_kafka" in str(e):
        raise SystemExit("Install: pip install 'cocoindex[kafka]'") from e
    raise

Prevention

Always install optional connectors via extras: cocoindex[kafka].
Declare the extra in pyproject.toml dependencies.
Smoke-test imports in CI with the same extras you deploy.

When it happens

Trigger: Any import of python/cocoindex/connectors/kafka/_source.py (directly or via the kafka connectors package __init__) when `from confluent_kafka import Message, TopicPartition` or `from confluent_kafka.aio import AIOConsumer` raises ImportError, i.e. confluent_kafka is absent or broken.

Common situations: Installing cocoindex without the [kafka] extra then importing the Kafka source connector; a slim/production Docker image that omitted extras; confluent_kafka failing to build (missing librdkafka) so the import fails even though installed.

Understand the failure class

Background: "X is not installed. Please install it with pip install Y": missing optional dependency errors — ImportError/ValueError raised when a library's optional extra was never installed — this error's family across 22 libraries.

Related errors

AI-assisted analysis of cocoindex-io/cocoindex@e84aa99b32 (2026-09-08). Data as JSON: /api/errors/1ec778c02b38a3f9. Report an issue: GitHub.