Docs/Kafka Connect

Kafka Connect SMT

Validate every Kafka record against a DataContractGate semantic contract in real-time — before it reaches your data warehouse or AI systems. Invalid records go to a dead-letter topic. Valid records continue unchanged.

Java 11+Kafka Connect 2.8+Apache 2.0v0.2.0

¶Quick Start

Three steps from zero to validated records.

Get your credentials

Sign up for a DataContractGate account, create a contract, and copy your API key and contract UUID from the Account page.

Install the connector

Build the Marketplace zip from the repo and extract it into your Connect plugin path:

mvn -f confluent-connector/pom.xml verify
unzip confluent-connector/target/datacontractgate-kafka-connect-contractgate-0.2.0.zip \
  -d /usr/share/confluent-hub-components/
# Restart Connect workers after installing

bash

Once listed on Confluent Marketplace:

confluent connect plugin install datacontractgate/kafka-connect-contractgate:0.2.0

bash

Add the SMT to your connector config

Add these lines to any existing connector's properties file:

transforms=contractgate
transforms.contractgate.type=io.datacontractgate.connect.smt.ContractGateValidator
transforms.contractgate.contractgate.api.url=https://api.datacontractgate.com
transforms.contractgate.contractgate.api.key=cg_live_YOUR_API_KEY
transforms.contractgate.contractgate.contract.id=YOUR_CONTRACT_UUID

# Route invalid records to a dead-letter topic
errors.deadletterqueue.topic.name=your-topic.dlq
errors.deadletterqueue.context.headers.enable=true

properties

That's it. Every record is now validated before reaching its destination.

¶Installation

Confluent Marketplace

After the listing is live, install with the Confluent CLI (the older confluent-hub client is deprecated on Confluent Platform 7.6+):

confluent connect plugin install datacontractgate/kafka-connect-contractgate:0.2.0

bash

Manual (Self-Managed Kafka)

Build the zip from github.com/nightmoose/contractgate and extract it into your plugin path:

# Typical plugin path locations:
# Confluent Platform: /usr/share/confluent-hub-components/
# Self-managed:       /opt/kafka/plugins/

unzip datacontractgate-kafka-connect-contractgate-0.2.0.zip -d /usr/share/confluent-hub-components/

# Add to connect-distributed.properties if not already:
plugin.path=/usr/share/confluent-hub-components

bash

Confluent Cloud (Custom Connector)

Upload the JAR via the Confluent Cloud UI under Connectors → Add plugin, then reference the SMT class in your connector config as shown above.

¶Configuration Reference

All settings are namespaced under contractgate.* to avoid conflicts with other SMTs in a chain.

Key	Default	Description
`contractgate.api.url`required	—	Base URL of the DataContractGate API. No trailing slash. Use `https://api.datacontractgate.com` for the hosted service.
`contractgate.contract.id`required	—	UUID of the contract to validate against. Copy from the Contracts page.
`contractgate.api.key`	""	Your API key (`x-api-key` header). Leave blank only for local dev with auth disabled.
`contractgate.contract.version`	"" (latest)	Pin to a specific contract version, e.g. `1.2.0`. Leave blank to always use the latest stable version — recommended for most pipelines. When set, the version is sent as the `?version=` query parameter on `POST /v1/ingest/{id}`.
`contractgate.on.failure`	DLQ	What to do when a record fails. `DLQ` — throw DataException for dead-letter routing. `TAG_AND_PASS` — add violation headers and pass through.
`contractgate.dry.run`	false	When `true`, validates without writing to the audit log. Useful for high-throughput pipelines where you want enforcement without DB write pressure.
`contractgate.connect.timeout.ms`	5000	TCP connection timeout to the DataContractGate API in milliseconds.
`contractgate.request.timeout.ms`	10000	Total HTTP request/response timeout in milliseconds. Keep well below Kafka Connect's task timeout.
`contractgate.add.result.headers`	true	Stamp `contractgate.*` metadata headers onto every record (pass or fail). See the headers reference below.
`contractgate.max.violation.headers`	5	Maximum number of individual violation detail headers to add. High violation counts can bloat headers — cap at a useful number.

¶Dead-Letter Queue Setup

When contractgate.on.failure=DLQ (the default), the SMT throws a DataException on validation failure. Kafka Connect's built-in error handling routes the original record to a dead-letter topic.

Full DLQ connector config

# ── DataContractGate SMT ──────────────────────────────────────
transforms=contractgate
transforms.contractgate.type=io.datacontractgate.connect.smt.ContractGateValidator
transforms.contractgate.contractgate.api.url=https://api.datacontractgate.com
transforms.contractgate.contractgate.api.key=cg_live_YOUR_API_KEY
transforms.contractgate.contractgate.contract.id=YOUR_CONTRACT_UUID
transforms.contractgate.contractgate.on.failure=DLQ

# ── Kafka Connect DLQ settings ─────────────────────────────
# The topic to route failed records to (create it first)
errors.deadletterqueue.topic.name=orders.dlq
errors.deadletterqueue.topic.replication.factor=3

# Include the full violation summary in DLQ record headers
errors.deadletterqueue.context.headers.enable=true

# Log every DLQ-routed record (set to false at high error rates)
errors.log.enable=true
errors.log.include.messages=true

# Retry transient errors before sending to DLQ
errors.retry.timeout=60000
errors.retry.delay.max.ms=5000

properties

Reading violation details from the DLQ

With errors.deadletterqueue.context.headers.enable=true, each DLQ record carries a header like:

__connect.errors.exception.message:
  DataContractGate validation failed — topic=orders partition=3 offset=1042
  contract=3fa85f64-... version=1.2.0
  2 violation(s): user_id [missing_required_field]: required field missing;
  amount [range_violation]: value -5 below minimum 0

text

¶Result Headers

When contractgate.add.result.headers=true (default), the following headers are added to every record — both passing and failing.

Header	Value
`contractgate.passed`	"true" or "false"
`contractgate.contract.version`	Resolved version string, e.g. "1.2.0"
`contractgate.violations.count`	Number of violations (0 on pass)
`contractgate.violation.0.field`	Dot-path of field, e.g. "customer.address.country"
`contractgate.violation.0.kind`	missing_required_field · type_mismatch · enum_violation · range_violation · pattern_mismatch · length_violation · undeclared_field
`contractgate.violation.0.message`	Human-readable explanation

Violation headers repeat for indices 0…N up to contractgate.max.violation.headers (default 5).

¶Tag & Pass Mode

Set contractgate.on.failure=TAG_AND_PASS to never drop records. Invalid records get violation headers and continue downstream — consumers can inspect contractgate.passed and decide what to do.

Use this when you want observability without enforcement — for example, shadowing a new contract version before promoting it to stable.

transforms.contractgate.contractgate.on.failure=TAG_AND_PASS
# Records now flow through regardless of violations.
# Downstream consumers can branch on contractgate.passed=false.

properties

¶Full Examples

S3 Sink with DLQ

{
  "name": "s3-sink-validated",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "4",
    "topics": "orders",
    "s3.region": "us-east-1",
    "s3.bucket.name": "my-data-lake",
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "format.class": "io.confluent.connect.s3.format.json.JsonFormat",

    "transforms": "contractgate",
    "transforms.contractgate.type": "io.datacontractgate.connect.smt.ContractGateValidator",
    "transforms.contractgate.contractgate.api.url": "https://api.datacontractgate.com",
    "transforms.contractgate.contractgate.api.key": "${file:/secrets/cg.properties:api.key}",
    "transforms.contractgate.contractgate.contract.id": "YOUR_CONTRACT_UUID",

    "errors.deadletterqueue.topic.name": "orders.dlq",
    "errors.deadletterqueue.context.headers.enable": "true",
    "errors.retry.timeout": "60000"
  }
}

json

JDBC Sink (high-throughput, dry-run audit)

{
  "name": "postgres-sink-validated",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "8",
    "topics": "user_events",
    "connection.url": "jdbc:postgresql://db:5432/analytics",

    "transforms": "contractgate",
    "transforms.contractgate.type": "io.datacontractgate.connect.smt.ContractGateValidator",
    "transforms.contractgate.contractgate.api.url": "https://api.datacontractgate.com",
    "transforms.contractgate.contractgate.api.key": "${file:/secrets/cg.properties:api.key}",
    "transforms.contractgate.contractgate.contract.id": "YOUR_CONTRACT_UUID",
    "transforms.contractgate.contractgate.dry.run": "true",
    "transforms.contractgate.contractgate.on.failure": "DLQ",

    "errors.deadletterqueue.topic.name": "user_events.dlq",
    "errors.deadletterqueue.context.headers.enable": "true"
  }
}

json

¶FAQ

Does this work with Avro / Protobuf / Schema Registry records?

Yes. The SMT converts any record value to JSON before sending to DataContractGate — Struct (Avro/Protobuf), Map, String, and byte[] are all handled automatically. Your schema registry setup is untouched.

What happens if the DataContractGate API is unreachable?

Network errors, HTTP 5xx, and auth/not-found statuses fail open — the SMT logs a warning and passes the record through so a transient outage does not halt the pipeline. HTTP 422 is not an outage: it means the event failed the contract, and the SMT applies DLQ or TAG_AND_PASS as configured.

Does it add latency to my pipeline?

The in-process engine, not the network hop, is what the repo measures. On the recorded host (2026-11-10, one thread, passing stripe-charges catalog samples) p99 was 0.93 µs and throughput was 1,547,368 events/sec. Reproduce with cargo run --release --bin contractgate-bench from a clone; see ops/bench/README.md. End-to-end latency also depends on the network between your Kafka brokers and the DataContractGate API, which that command does not include.

Can I use it with Confluent Cloud managed connectors?

Yes — upload the JAR as a custom connector plugin via the Confluent Cloud UI, then reference the SMT class in your connector config as shown in the examples above.

How do I pin a contract version?

Set contractgate.contract.version=1.2.0. The connector sends this as the ?version= query parameter on POST /v1/ingest/{id}. Leave it blank (the default) to always resolve to the latest stable version — this lets you promote new contract versions without redeploying connectors.

Can I chain this with other SMTs?

Yes. List multiple transforms: transforms=contractgate,maskPii,routeByField. The DataContractGate SMT works anywhere in the chain; put it first to validate raw records before any transforms mutate them.

Ready to get started?

Create a free account to get your API key and contract UUID.