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 target connector module fails because confluent_kafka (specifically confluent_kafka.aio.AIOProducer) is not installed. The module wraps the import in a try/except and re-raises an ImportError telling the user to install the cocoindex[kafka] extra.

Source

Thrown at python/cocoindex/connectors/kafka/_target.py:18

"""
Kafka target for CocoIndex.

This module provides a two-level target state system for Kafka:
1. Topic level: Lightweight container for generation tracking (user-managed topic)
2. Message level: Produces messages to the topic for upserts/deletes
"""

from __future__ import annotations

import asyncio
from dataclasses import dataclass
from typing import Callable, Collection, Generic, NamedTuple, Sequence

try:
    from confluent_kafka.aio import AIOProducer  # 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

import cocoindex as coco
from cocoindex.connectorkits.fingerprint import fingerprint_bytes, fingerprint_str
from cocoindex._internal.context_keys import ContextKey, ContextProvider
from cocoindex._internal.datatype import TypeChecker

# --- Type aliases ---

_MessageFingerprint = bytes

# --- Internal types ---


class _TopicKey(NamedTuple):
    producer_key: str

View on GitHub (pinned to e84aa99b32)

Solutions

Install the extra: `pip install 'cocoindex[kafka]'`.
If confluent_kafka is installed but the import still fails, upgrade it (`pip install -U confluent_kafka`) — the aio submodule requires a recent version.
Verify with `python -c "from confluent_kafka.aio import AIOProducer"` before using the connector.

Example fix

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

Defensive patterns

Strategy: try-catch

Validate before calling

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

Try / catch

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

Prevention

Install cocoindex[kafka] explicitly in every environment.
Ensure confluent_kafka is recent enough to include confluent_kafka.aio.
Verify the import in a CI smoke test.

When it happens

Trigger: Any import of python/cocoindex/connectors/kafka/_target.py (directly or via the kafka connectors package) when `from confluent_kafka.aio import AIOProducer` raises ImportError.

Common situations: Base cocoindex install without extras; environments where an older confluent_kafka version lacks confluent_kafka.aio; wheels unavailable on the platform causing install (and later import) failure.

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/70fdabfc7853d0eb. Report an issue: GitHub.