Why One Kafka Connector Should Replace 200: How AI Is Finally Fixing the Integration Sprawl Problem

Today we’re open-sourcing kafka-connect-ai — a single, Apache 2.0 licensed Kafka Connect connector that replaces hundreds of single-purpose connectors with 12 protocol adapters and an LLM-powered transformation pipeline. 

The technical breakthroughs that make this possible: compiled transforms that let the LLM generate reusable transformation code (declarative mappings at ~500ns/record, sandboxed GraalVM JavaScript at ~2μs/record) instead of calling the model per-record; 4-tier model routing that automatically sends simple records to the cheapest capable model — or bypasses the LLM entirely for deterministic transforms; semantic caching via Redis vector similarity that eliminates duplicate LLM calls; and PII masking that strips sensitive fields before they ever reach the model. The result: a universal connector that handles HTTP, JDBC, MongoDB, gRPC, Kafka-to-Kafka, Redis, Cassandra, Kinesis, cloud data warehouses, WebSocket/SSE streaming, LDAP, and SAP — with LLM costs that converge toward zero as the system learns your schemas.

Each adapter covers an entire category of systems. The HTTP adapter covers every REST API. The JDBC adapter covers every SQL database. The gRPC adapter covers any gRPC service. Adding a new integration is a configuration change, not a development project.

Architecture: How It Actually Works

kafka-connect-ai is not an AI wrapper on top of existing connectors. It is a purpose-built transformation pipeline where the LLM generates code, not just data — and that code is cached and reused across all subsequent records with the same schema shape.

The Key Innovation: Compiled Transforms

This is the breakthrough that makes AI-powered data transformation economically viable at scale.

Traditional approaches call the LLM for every record. kafka-connect-ai takes a fundamentally different approach: the LLM generates transformation code, not transformed data. That code is compiled, validated, cached, and reused for every subsequent record with the same schema shape.

The two-tier execution strategy:

• Tier 0 — Declarative mappings (~500ns/record): Pure Java, zero security risk. Handles field renames, JSONPath mappings, defaults, and type casts. No scripting engine involved.
• Tier 1 — Sandboxed JavaScript (~2μs/record after JIT): GraalVM Polyglot with full sandboxing — no file system access, no network access, no host access. Configurable memory limit (10MB default) and execution timeout (100ms default). Handles complex transformations with conditional logic.

4-Tier Model Routing

Not every record needs the same model. kafka-connect-ai automatically routes records to the cheapest capable tier:

Records cascade down tiers — the router picks the cheapest model that can handle the complexity.

Cost Optimisation Stack

Four layers work together to minimise LLM spend:

Privacy: PII Masking

Sensitive fields are masked before data reaches the LLM and restored after transformation:

Input    {"name": "John Smith", "email": "[email protected]", "order_id": 42}
  ↓ mask
To LLM   {"name": "[MASKED_0]", "email": "[MASKED_1]", "order_id": 42}
  ↓ transform
From LLM  {"customer": "[MASKED_0]", "contact": "[MASKED_1]", "id": 42}
  ↓ unmask
Output   {"customer": "John Smith", "contact": "[email protected]", "id": 42}

Configure by field name (case-insensitive) or regex pattern matching. Meets GDPR requirements by ensuring PII never leaves your infrastructure boundary to reach the LLM provider.

Production Observability: 16 JMX Metrics

Every stage of the pipeline is instrumented via Micrometer:

All prefixed with connect.ai.* and exportable to Prometheus, Datadog, Grafana, or any Micrometer-compatible backend.

git clone https://github.com/osodevops/kafka-connect-ai.git
cd kafka-connect-ai
mvn clean package -pl kafka-connect-ai-connect -am -DskipTests
cd docker && docker compose up -d

This starts Kafka (KRaft mode), Schema Registry, Kafka Connect with kafka-connect-ai pre-loaded, PostgreSQL, and Redis.

2. Verify the Connector Plugin is Loaded

curl -s http://localhost:8083/connector-plugins | jq '.[].class' | grep ai

You should see:

"sh.oso.connect.ai.connect.source.AiSourceConnector"
"sh.oso.connect.ai.connect.sink.AiSinkConnector"

3. Example: REST API Source with LLM Transformation

Pull data from any REST API, transform it with AI, and write to Kafka:

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "weather-api-source",
    "config": {
      "connector.class": "sh.oso.connect.ai.connect.source.AiSourceConnector",
      "tasks.max": "1",

      "connect.ai.source.adapter": "http",
      "connect.ai.topic": "weather-events",

      "http.source.url": "https://api.openweathermap.org/data/2.5/weather?q=London&appid=YOUR_KEY",
      "http.source.poll.interval.ms": "300000",
      "http.source.auth.type": "none",

      "ai.llm.provider": "anthropic",
      "ai.llm.api.key": "${ANTHROPIC_API_KEY}",
      "ai.llm.model": "claude-sonnet-4-20250514",
      "ai.agent.system.prompt": "Extract: city, temp_celsius (convert from Kelvin), humidity_pct, wind_speed_ms, conditions, recorded_at (ISO 8601)."
    }
  }'

4. Example: Database CDC with PII Masking

Capture changes from PostgreSQL with sensitive fields automatically masked before reaching the LLM:

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "orders-db-source",
    "config": {
      "connector.class": "sh.oso.connect.ai.connect.source.AiSourceConnector",
      "tasks.max": "1",

      "connect.ai.source.adapter": "jdbc",
      "connect.ai.topic": "order-events",

      "jdbc.url": "jdbc:postgresql://postgres:5432/kcai",
      "jdbc.user": "kcai",
      "jdbc.password": "kcai",
      "jdbc.table": "orders",
      "jdbc.query.mode": "timestamp",
      "jdbc.timestamp.column": "updated_at",

      "ai.llm.provider": "anthropic",
      "ai.llm.api.key": "${ANTHROPIC_API_KEY}",
      "ai.llm.model": "claude-sonnet-4-20250514",
      "ai.agent.system.prompt": "Normalise order rows into events with: order_id, customer_name, amount_usd, currency, status, event_timestamp.",

      "ai.privacy.mask.fields": "email,phone,ssn,credit_card",
      "ai.privacy.mask.patterns": "\\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Z]{2,}\\b",
      "ai.privacy.unmask.output": "true"
    }
  }'

5. Example: Kafka-to-Kafka Transformation with Full Cost Optimisation

Consume from one cluster, transform with AI, produce to another — with all cost-saving layers enabled:

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "k2k-normaliser",
    "config": {
      "connector.class": "sh.oso.connect.ai.connect.source.AiSourceConnector",
      "tasks.max": "1",

      "connect.ai.source.adapter": "kafka",
      "connect.ai.topic": "normalised-events",

      "kafka.source.bootstrap.servers": "upstream-kafka:9092",
      "kafka.source.topics": "raw-events",

      "ai.llm.provider": "anthropic",
      "ai.llm.api.key": "${ANTHROPIC_API_KEY}",
      "ai.llm.model": "claude-sonnet-4-20250514",
      "ai.agent.system.prompt": "Merge legacy event formats into a unified schema: event_id, event_type, source_system, payload (object), occurred_at (ISO 8601).",
      "ai.agent.enable.prompt.caching": "true",

      "ai.router.enabled": "true",
      "ai.router.deterministic.patterns": "field_rename,type_cast,timestamp_format",
      "ai.llm.model.fast": "claude-haiku-4-5-20251001",

      "ai.cache.enabled": "true",
      "ai.cache.redis.url": "redis://redis:6379",

      "ai.compiled.transform.enabled": "true",
      "ai.compiled.transform.ttl.seconds": "86400",

      "ai.batch.size": "50",
      "ai.batch.wait.ms": "5000",
      "ai.batch.parallel.calls": "4",

      "ai.llm.rate.limit": "100"
    }
  }'

6. Example: MongoDB to Snowflake via Kafka (Sink Connector)

Write transformed records from Kafka to a cloud data warehouse:

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "warehouse-sink",
    "config": {
      "connector.class": "sh.oso.connect.ai.connect.sink.AiSinkConnector",
      "tasks.max": "2",
      "topics": "normalised-events",

      "connect.ai.sink.adapter": "warehouse",

      "warehouse.type": "snowflake",
      "warehouse.snowflake.url": "https://acme.snowflakecomputing.com",
      "warehouse.snowflake.database": "ANALYTICS",
      "warehouse.snowflake.schema": "PUBLIC",
      "warehouse.snowflake.table": "EVENTS",

      "ai.llm.provider": "anthropic",
      "ai.llm.api.key": "${ANTHROPIC_API_KEY}",
      "ai.llm.model": "claude-sonnet-4-20250514",
      "ai.agent.system.prompt": "Map event fields to warehouse columns: EVENT_ID, EVENT_TYPE, SOURCE, PAYLOAD_JSON, CREATED_AT."
    }
  }'

Comparison