Kafka Connector

The Kafka connector streams document chunks to Apache Kafka topics as individual messages. This connector is target-only and chunks-only - only produces chunked output suitable for downstream processing.

Prerequisites

• Kafka Cluster: Access to an Apache Kafka cluster (self-hosted, Confluent Cloud, AWS MSK, etc.)
• Kafka Topic: A topic to publish chunks to
• Bootstrap Servers: Kafka broker addresses
• Authentication (if required): SASL credentials and potentially CA Cert for secure clusters

Setup and Authentication

Kafka Cluster Access

Ensure you have:

1. Bootstrap server addresses (e.g., broker1:9092,broker2:9092)
2. Topic name where chunks will be published
3. Authentication credentials (if your cluster requires SASL)

Authentication Methods

SASL Authentication

For production clusters (Confluent Cloud, secure on-premise):

• Supported mechanisms: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512
• Provide username and password
• Uses SASL_SSL by default (TLS + SASL)

Custom CA Certificate

If your cluster uses a self-signed or internal CA certificate:

1. Obtain the CA certificate file (.crt or .pem)
2. Base64-encode it: cat your_ca.crt | base64 | tr -d '\n'
3. Include the encoded string in the ca_cert field of your auth configuration

Configuration

The Kafka connector is used as a target in the Batch API.

Required Parameters

Optional Parameters

SASL Auth Object

When providing auth, include these fields:

Connector-Specific Behavior

Message Structure

Each chunk is published as a separate Kafka message with:

• Key: Based on key_mode (doc_id, chunk_id, or null)
• Value: JSON object containing chunk text and metadata
• Headers: Include chunk_id for deduplication

Example message value:

{
  "text": "This is the chunk text content...",
  "metadata": {
    "source": "document.pdf",
    "page": 1
  },
  "doc_id": "abc123",
  "chunk_index": 0,
  "chunk_id": "def456",
  "page_numbers": [1],
  "headings": ["Introduction"]
}

Key Modes

doc_id (default, recommended):

• All chunks of a document go to the same partition
• Ordered delivery within a document (requires acks: "all")
• Warning: Do NOT enable log compaction with this mode - it would keep only the last chunk per document

chunk_id:

• Content-addressed stable hash
• Chunks spread across partitions for parallelism
• No ordering guarantee

none:

• Null key, round-robin distribution
• No ordering, maximum throughput

Acknowledgment Modes

all (default, most durable):

• Wait for all in-sync replicas
• Enables idempotent producer (prevents duplicates on retry)
• Required for ordered delivery with key_mode: doc_id

1 (leader ack):

• Wait for leader replica only
• Faster but less durable

0 (fire-and-forget):

• No acknowledgment
• Fastest but may lose messages

Chunking

Documents are automatically chunked using Docling's chunking engine. Control chunking behavior with options.chunking_options in your batch request.

Note that this field is optional. Docling will choose a default tokenizer for you - it just may not align with downstream processing.

{
  "options": {
    "chunking_options": {
      "chunker": "hybrid",
      "tokenizer": "sentence-transformers/all-MiniLM-L6-v2",
      "max_tokens": 512
    }
  }
}

Security and Permissions

Kafka ACLs

Your SASL user needs permissions for:

• WRITE on the target topic
• CREATE on the topic (if it doesn't exist and broker allows auto-creation)

TLS/SSL

For secure clusters:

• Use SASL_SSL or SSL protocol
• Broker certificates must be valid or ca_cert must be provided
• Set verify_certs: false only for development with self-signed certs

Limitations

• Target Only: Kafka connector cannot read documents (source mode not supported)
• Chunks Only: Only publishes chunked output, not full documents
• Message Size: Kafka has a default max message size of 1 MB. Ensure max_tokens in chunking options produces chunks within this limit
• Ordering: Only guaranteed within a partition when using key_mode: doc_id with acks: all
• No Transaction Support: Messages are published independently, not in Kafka transactions
• Log Compaction Warning: Do not enable log compaction with key_mode: doc_id - it would silently discard all but the last chunk

Usage Examples

There are three main ways to interface with the connectors. All use the same underlying POST /v1/convert/source/batch endpoint.

Tasks UI

Navigate to the Tasks view and select "Create Task +". Select Batch as the task type (connectors use batch tasks, not single).

Fill in the fields as prompted. They should correspond to the fields gathered above (excluding 'kind').

Screenshots for this connector will be added here.

REST API

curl -X POST "${DOCLING_SERVICE_URL}/v1/convert/source/batch" \
  -H "X-Api-Key: ${DOCLING_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{
    "sources": [
      {
        "kind": "s3",
        "endpoint": "s3.us-east-1.amazonaws.com",
        "access_key": "YOUR_ACCESS_KEY",
        "secret_key": "YOUR_SECRET_KEY",
        "bucket": "my-documents",
        "key_prefix": "pdfs/"
      }
    ],
    "target": {
      "kind": "kafka_chunks",
      "bootstrap_servers": ["broker1.example.com:9092", "broker2.example.com:9092"],
      "topic": "docling.chunks",
      "key_mode": "doc_id",
      "auth": {
        "kind": "sasl",
        "mechanism": "PLAIN",
        "username": "your-username",
        "password": "your-password"
      },
      "acks": "all",
      "compression_type": "lz4"
    },
    "options": {
      "to_formats": ["json"],
      "chunking_options": {
        "chunker": "hybrid",
        "tokenizer": "sentence-transformers/all-MiniLM-L6-v2",
        "max_tokens": 512
      }
    }
  }'

With Custom CA Certificate

curl -X POST "${DOCLING_SERVICE_URL}/v1/convert/source/batch" \
  -H "X-Api-Key: ${DOCLING_API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{
    "sources": [
      {
        "kind": "http",
        "url": "https://example.com/document.pdf"
      }
    ],
    "target": {
      "kind": "kafka_chunks",
      "bootstrap_servers": ["internal-broker.corp:9092"],
      "topic": "docling.chunks",
      "auth": {
        "kind": "sasl",
        "mechanism": "PLAIN",
        "username": "your-username",
        "password": "your-password",
        "ca_cert": "xXx...BASE64_ENCODED_CERT...xXx"
      }
    },
    "options": {
      "chunking_options": {
        "chunker": "hybrid",
        "max_tokens": 512
      }
    }
  }'

Python SDK

from docling.service_client import DoclingServiceClient
from docling.datamodel.service.requests import S3SourceRequest
from docling.datamodel.service.targets import GenericTargetRequest
import os

SERVICE_URL = os.getenv("DOCLING_SERVICE_URL")
API_KEY = os.getenv("DOCLING_API_KEY")

# Kafka target with SASL authentication
target = GenericTargetRequest(
    kind="kafka_chunks",
    bootstrap_servers=["broker1.example.com:9092", "broker2.example.com:9092"],
    topic="docling.chunks",
    key_mode="doc_id",
    auth={
        "kind": "sasl",
        "mechanism": "PLAIN",
        "username": "your-username",
        "password": "your-password"
    },
    acks="all",
    compression_type="lz4"
)

with DoclingServiceClient(url=SERVICE_URL, api_key=API_KEY) as client:
    job = client.submit_batch(
        sources=[
            S3SourceRequest(
                endpoint="s3.us-east-1.amazonaws.com",
                access_key="YOUR_ACCESS_KEY",
                secret_key="YOUR_SECRET_KEY",
                bucket="my-documents",
                key_prefix="pdfs/"
            )
        ],
        target=target,
        output_formats=["json"],
        chunking_options={
            "chunker": "hybrid",
            "tokenizer": "sentence-transformers/all-MiniLM-L6-v2",
            "max_tokens": 512
        }
    )
    
    # Wait for completion
    response = job.result()
    print(f"Processed {response.num_converted} documents")
    print(f"Succeeded: {response.num_succeeded}")
    print(f"Failed: {response.num_failed}")

Related Documentation

• Batch API Reference - Complete batch endpoint documentation
• Connectors Overview - All available connectors
• Apache Kafka Documentation - Official Kafka documentation
• Confluent Cloud - Managed Kafka service documentation