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

Updated November 2026: We have restructured this March 2026 launch post and checked it against the current kafka-connect-ai code. We corrected the batch and Snowflake sink configuration keys, updated the metrics list, replaced fixed model prices with relative costs, and clarified how Debezium compares.

In March 2026 we open-sourced 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.

Four ideas make it practical. Compiled transforms let the LLM generate reusable transformation code (declarative mappings at about 500 ns per record, sandboxed GraalVM JavaScript at about 2 μs per record) instead of calling the model for every record. 4-tier model routing sends simple records to the cheapest capable model, or bypasses the LLM entirely for deterministic transforms. Semantic caching with Redis vector similarity removes duplicate LLM calls. PII masking strips sensitive fields before they reach the model.

What is the Kafka connector sprawl problem?

Ask any Kafka engineer what keeps them up at night, and it is rarely the broker. It is Connect: the ever-expanding sprawl of connector JARs, configuration models, dependency conflicts and version mismatches that builds up in every mature Kafka deployment until it becomes a full-time job.

200 connectors, one fundamental operation

OSO engineers regularly walk into organisations running fifteen to twenty different connector JARs in a single Kafka Connect cluster. Each has its own version, its own transitive dependencies and its own classpath conflicts with the others. Maintaining that ecosystem (updating JARs, resolving dependency collisions, debugging serialisation mismatches) can take two to three engineers who should be building pipelines.

Every connector also brings a new mental model. The JDBC connector structures its configuration differently from the HTTP connector. The S3 connector has its own authentication surface. The Elasticsearch connector has its own idea of index routing. There is no unified configuration language, so knowledge does not transfer: expertise with the Salesforce connector gives an engineer nothing useful when diagnosing the MongoDB connector.

Licensing has turned sprawl into a strategic risk

Apache Kafka itself remains Apache 2.0, but many connectors have followed a different path. The Confluent Community License, introduced in 2018, covers Schema Registry, ksqlDB and a significant part of the connector catalogue, and prohibits offering those components as competing SaaS products. Premium connectors go further: mission-critical integrations such as Oracle CDC sit behind enterprise subscriptions.

The risks are concrete:

• Licensing terms on connectors you use today may change.
• Costs rise once you are locked into a platform.
• Moving away from connector-specific investments is slow and expensive.
• Innovation may slow as the contributing organisation’s priorities shift.

kafka-connect-ai is Apache 2.0 licensed: no community licence, no premium tier and no enterprise restrictions on the open source connector.

Why build Kafka connectors around protocols, not products?

After more than thirty enterprise Kafka engagements, OSO engineers reached a conclusion that runs against the way the Connect ecosystem grew: the connector-per-system model is a category error. The right abstraction is the protocol.

What changes between Salesforce and HubSpot is not the protocol; both speak REST over HTTP. What changes between PostgreSQL and MySQL is not the protocol; both speak JDBC. What changes is the configuration surface and the shape of the data. Build a clean implementation of each protocol once, and let AI handle the per-system transformation. kafka-connect-ai ships with 12 protocol adapters:

Each adapter covers a whole category of systems. Adding a new integration within a category is a configuration change, not a development project.

How do compiled transforms make AI transformation affordable?

kafka-connect-ai is not an AI wrapper on top of existing connectors. It is a purpose-built transformation pipeline in which the LLM generates code, not just data, and that code is cached and reused for every later record with the same schema shape.

Compiled transforms

Traditional approaches call the LLM for every record. kafka-connect-ai takes a different approach: the LLM generates transformation code, not transformed data. That code is compiled, validated against sample records, cached by schema fingerprint and reused.

There are two execution tiers for compiled code:

• Declarative mappings (about 500 ns per record): pure Java with no scripting engine. Handles field renames, JSONPath mappings, defaults and type casts.
• Sandboxed JavaScript (about 2 μs per record after JIT warm-up): GraalVM Polyglot with no file system, network or host access, a configurable memory limit (10 MB by default) and execution timeout (100 ms by default). Handles complex transformations with conditional logic.

4-tier model routing

Not every record needs the same model. With routing enabled, kafka-connect-ai sends each batch to the cheapest tier that can handle it:

Model prices change often, so check your provider’s current pricing; the connector tracks spend per call in its cost metric.

Cost optimisation stack

Five layers work together to minimise LLM spend:

How does kafka-connect-ai protect data and stay observable?

PII masking

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

Input     {"name": "John Smith", "email": "john@example.com", "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": "john@example.com", "id": 42}

Configure masking by field name (case-insensitive) or by regex pattern. Because masked values never leave your infrastructure, it supports GDPR data-minimisation requirements when you use a hosted LLM provider.

Production observability

Every stage of the pipeline is instrumented with Micrometer and exposed over JMX:

All metrics are prefixed with connect.ai and can be exported to Prometheus, Datadog, Grafana or any Micrometer-compatible backend.

How do you get started with kafka-connect-ai?

You need Java 17 or later, Maven 3.9 or later, Docker for the quick-start stack, and an Anthropic or OpenAI API key.

1. Start the stack

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 in KRaft mode, Schema Registry, Kafka Connect with kafka-connect-ai pre-loaded, PostgreSQL and Redis.

2. Check 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. REST API source with LLM transformation

Pull data from a REST API, transform it with AI and write it 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. Database polling source with PII masking

Capture changed rows from PostgreSQL with sensitive fields masked before they reach 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. Kafka-to-Kafka transformation with every cost layer on

Consume from one cluster, transform with AI and 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.max.wait.ms": "5000",
      "ai.batch.parallel.calls": "4",
      "ai.llm.rate.limit": "100"
    }
  }'

6. Kafka to Snowflake with the warehouse sink

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.provider": "snowflake",
      "snowflake.url": "https://acme.snowflakecomputing.com",
      "snowflake.user": "KAFKA_LOADER",
      "snowflake.private.key": "${SNOWFLAKE_PRIVATE_KEY}",
      "snowflake.warehouse": "LOAD_WH",
      "snowflake.database": "ANALYTICS",
      "snowflake.schema": "PUBLIC",
      "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."
    }
  }'

How do you adopt kafka-connect-ai without a forklift migration?

kafka-connect-ai is a standard Kafka Connect connector. It runs inside your existing Connect cluster, alongside your existing connectors, and installs like any other connector JAR. There is no forklift migration.

Start with phase 1, where the operational pain is greatest. In phase 4, kafka-connect-ai being Apache 2.0 means you can run it on-premises, in the cloud or as a service without licensing constraints. For managed Connect on AWS, see our MSK Connect guide.

How does kafka-connect-ai compare with other Kafka connectors?

When should you not use kafka-connect-ai?

• Sub-millisecond latency: the first LLM call for each new schema shape adds latency (100 ms to 5 s). Compiled transforms bring later records down to microseconds, but that first call is unavoidable.
• Full CDC from the write-ahead log: for database replication with transaction ordering, Debezium is purpose-built. kafka-connect-ai’s JDBC adapter uses polling queries.
• Binary data: kafka-connect-ai works with JSON. For Avro, Protobuf or binary payloads, use specialised connectors.
• Purely structural transforms: if all you need is renames and type casts, Kafka Connect SMTs are simpler, although Tier 0 handles them too.

What should you do next?

The connector-per-system model was a pragmatic answer to a 2015 problem. Today it is a source of operational overhead, licensing risk and organisational fragility: hundreds of connectors, each encoding the same structural logic for a different product, each needing specialist knowledge. Protocols matter; products do not. What was missing was a transformation engine flexible enough to handle per-system differences without a separate codebase for each system, and smart enough to need the LLM once per schema shape rather than once per record.

• Clone kafka-connect-ai and run the quick-start stack.
• Pick one high-friction HTTP or JDBC connector and run kafka-connect-ai alongside it.
• Turn on routing, compiled transforms and caching, and watch the cost and router metrics.
• Audit which of your connectors carry licensing or subscription exposure.

The project is open source, the adapter interface is designed for community contribution, and we welcome issues and pull requests. If you want help planning a connector consolidation, talk to OSO’s Kafka engineers.

Key takeaway: Kafka connectors mostly repeat the same work over a few protocols. Consolidate them into protocol adapters, let the LLM write transformation code once per schema shape, and route, cache and mask so AI-powered integration stays cheap and safe.

Frequently asked questions