Schema Registry & Connect Ecosystem

Schema Registry and Kafka Connect are the two components that turn a raw Kafka cluster into a governed integration platform. Schema Registry enforces a data contract between producers and consumers so that a field-type change is rejected at registration time instead of silently breaking deserialization downstream. Kafka Connect moves data in and out of Kafka through a pluggable connector model that handles offset tracking, fault tolerance, and parallelism for you.

This page covers the architecture and the production gotchas that matter when you run both in anger: which compatibility modes actually protect you (and the upgrade order each one assumes), the exact and frequently mis-documented connector configuration keys, how task-failure metrics misbehave under JMX, and what replicating Schema Registry across data centers really requires.

The two authoritative references are the Confluent Schema Registry documentation and the Apache Kafka Connect documentation. Where this page cites a specific behavior, it links the page that documents it.

The Role of Schema Registry in Data Governance

Schema Registry stores and retrieves versioned schemas for Kafka data, but its operational value is the enforcement point it gives you. When a producer serializes a message with one of the Schema Registry-aware serializers (Avro, Protobuf, or JSON Schema), the serializer registers the schema under a subject and the registry checks it against that subject’s configured compatibility level. If the change violates the rule — for example, removing a required field under BACKWARD compatibility — registration fails and the produce call throws. Validation moves left, from consumers (where a bad record is discovered too late) to the produce path (where it surfaces in CI or at deploy time).

Schema Registry stores its state in a single compacted Kafka topic, _schemas, so it inherits Kafka’s replication and durability and needs no external database. The registry process itself is effectively stateless: every instance is a consumer of _schemas and the group elects one primary for writes, which is why you can run several instances behind a load balancer. Each registered schema gets a globally unique integer ID. The serializer writes a wire-format prefix ahead of the payload — a magic byte (0x00) followed by the 4-byte schema ID as a big-endian int — and the consumer’s deserializer reads that ID to fetch the exact writer schema. That is what lets a consumer decode a record written with an older schema version: the bytes on the wire always declare which schema produced them.

Schema Subjects and Naming Strategies

Schemas are organized into subjects. The default TopicNameStrategy derives the subject from the topic name plus the record part: for topic orders, the value subject is orders-value and the key subject is orders-key. This is simple but ties one subject to one topic, which forbids putting more than one event type on a topic.

RecordNameStrategy keys the subject on the Avro/Protobuf fully qualified record name, and TopicRecordNameStrategy combines topic and record name. Both let a single logical type — say a Customer record — be reused across topics, and both let you put multiple event types in one topic. You set them per serializer with key.subject.name.strategy / value.subject.name.strategy. The catch: once you move off TopicNameStrategy, a consumer subscribed to a multi-type topic must be able to deserialize every type it may encounter, so the choice constrains your consumer code, not just your topic layout. Keep the strategy consistent across environments when Managing Schemas Across Multiple Environments.

Compatibility Types in Practice

A subject’s compatibility level governs how its schema may evolve. The mechanically allowed changes are the same across the common modes — add a field with a default, or remove a field that had a default (Avro/Protobuf call these “optional”). What differs is the direction the registry checks and, consequently, the order in which you must roll out producers and consumers:

Each mode has a *_TRANSITIVE variant: the non-transitive check is against the latest registered version only, while the transitive variant checks against all prior versions. Pick transitive when consumers may still be reading data written under any historical schema (e.g., long-retention or replayable topics); the non-transitive default is cheaper but only guarantees the most recent hop. The full rule matrix is in the Schema Evolution and Compatibility reference and is covered operationally in Schema Evolution and Compatibility in Confluent Schema Registry.

The most common production incident here is adding a field without a default under BACKWARD: registration fails at deploy time, which is the system working as intended — but teams often misread it as a registry outage. The fix is always in the schema (give the field a default), never a compatibility downgrade.

Set the compatibility level with the registry REST API — globally, or per subject (a subject-level setting overrides the global one):

# Global default
curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"compatibility": "BACKWARD"}' \
  http://schema-registry:8081/config

# Override for one subject
curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"compatibility": "FULL"}' \
  http://schema-registry:8081/config/orders-value

Kafka Connect Architecture and Deployment Modes

Kafka Connect streams data between Kafka and external systems through pluggable connectors. It handles the parts you would otherwise reimplement per integration: offset tracking, restart-from-offset, retries, and distributing work across processes.

Connect separates connectors (the high-level definition of a data-movement job) from tasks (the units that run in parallel). A source connector decides how to split the work — for a JDBC source, how to spread tables across tasks — and the framework schedules those tasks across the worker cluster. Task count is bounded by the connector’s tasks.max and by how finely the connector can partition its work: a single-table JDBC source cannot exceed one task no matter how high you set tasks.max. See Kafka Connect Architecture and Deployment Modes for the worker/task model in detail.

Distributed Mode Configuration

In production Connect runs in distributed mode: several worker processes share a group.id and coordinate through Kafka’s consumer-group membership protocol, so a failed worker’s tasks are rebalanced onto survivors. Configuration, offsets, and status live in three internal compacted topics that all workers in the group share.

# connect-distributed.properties
bootstrap.servers=kafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092
group.id=connect-cluster-prod
key.converter=io.confluent.connect.avro.AvroConverter
value.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://schema-registry:8081
value.converter.schema.registry.url=http://schema-registry:8081
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status
config.storage.replication.factor=3
offset.storage.replication.factor=3
status.storage.replication.factor=3
plugin.path=/usr/share/java/connect-plugins

Two failure modes worth naming: group.id collisions and topic mismatches. Two Connect clusters that accidentally share a group.id will fight over the same internal topics and corrupt each other’s state, so the group.id must be unique per logical cluster. And config.storage.topic must be a single-partition compacted topic — if Connect auto-creates it with multiple partitions or non-compacted cleanup, config and offset recovery break silently after a restart.

The converter choice is what wires Connect into your governance model. With AvroConverter (or the Protobuf / JSON Schema converters) pointed at Schema Registry, source connectors auto-register the schema for every topic they write, and sink connectors validate on read. Teams using other formats can swap in io.confluent.connect.protobuf.ProtobufConverter or io.confluent.connect.json.JsonSchemaConverter — see Using Protobuf with Kafka and Schema Registry and JSON Schema Support in Confluent Platform.

Single Message Transforms

Single Message Transforms (SMTs) are per-record functions chained inside a connector — useful for renaming, routing, masking, or dropping fields without a separate stream-processing job. A transform type targets either the key or the value via its $Key / $Value inner class. SMTs run in the order listed in transforms, so place a router before any transform that depends on the final topic name.

{
  "name": "jdbc-source-orders",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "transforms": "renameTopic,dropSensitive",
    "transforms.renameTopic.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.renameTopic.regex": "mysql-orders-(.*)",
    "transforms.renameTopic.replacement": "orders-$1",
    "transforms.dropSensitive.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
    "transforms.dropSensitive.exclude": "credit_card_number,social_security"
  }
}

ReplaceField uses exclude (drop these fields) and include (keep only these). The older blacklist / whitelist names still work but are deprecated — prefer exclude / include, per the ReplaceField reference. Note that dropping a field with ReplaceField changes the record’s schema, which the converter then registers as a new schema version — so a masking SMT can itself trip a compatibility rule on the destination subject.

Change Data Capture with Debezium

Change Data Capture streams row-level changes out of a database’s transaction log and into Kafka, enabling replication, cache invalidation, and event-driven downstreams without touching application code. Debezium is built on Kafka Connect and registers a schema per table with Schema Registry, so every CDC stream carries a versioned contract. The pattern is covered in Change Data Capture with Debezium and Kafka.

Deploying a Debezium MySQL Connector

The configuration below captures inventory.customers and inventory.orders. Note the property names: Debezium 2.x renamed the server-name and schema-history keys, and the older database.server.name / database.history.kafka.* forms used in many tutorials no longer apply.

{
  "name": "mysql-inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql-primary",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "${vault:secret/mysql-debezium:password}",
    "database.server.id": "184054",
    "topic.prefix": "mysql-inventory",
    "database.include.list": "inventory",
    "table.include.list": "inventory.customers,inventory.orders",
    "schema.history.internal.kafka.bootstrap.servers": "kafka-broker-1:9092,kafka-broker-2:9092",
    "schema.history.internal.kafka.topic": "schema-changes.inventory",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false"
  }
}

• topic.prefix (formerly database.server.name) is the logical name and the prefix for every topic this connector writes; database.server.id is a unique numeric replica ID for the binlog client. Two MySQL connectors sharing a database.server.id will be disconnected by the server as duplicate replicas — a subtle multi-connector gotcha.
• ExtractNewRecordState (the unwrap SMT) flattens Debezium’s before/after envelope to just the after-state. Keep drop.tombstones=false so deletes still emit a null-value tombstone for log-compacted sinks; setting it to true makes deletes vanish, which silently breaks compaction-based mirrors of the source table.

Building Custom Kafka Connect Connectors

When you have a proprietary system without a pre-built connector, the Connect SDK lets you implement just the data-movement logic while the framework handles distribution, retries, and offset commits. A connector is a Connector subclass (config + task splitting) plus a Task subclass (the actual transfer). See Building Custom Kafka Connect Connectors for testing and packaging.

Source Task Skeleton

A SourceTask returns SourceRecord objects built from Connect’s runtime Schema and Struct types. You do not serialize to Avro yourself — the worker applies the configured value.converter (here AvroConverter) to the record, which is what registers the schema and writes the bytes. Building your own byte[] in poll() would bypass the converter entirely, so the record would never reach Schema Registry and downstream Avro consumers would fail to deserialize it.

public class LegacySourceTask extends SourceTask {
    private LegacyClient client;
    private String topic;

    private static final Schema VALUE_SCHEMA = SchemaBuilder.struct()
        .name("com.example.LegacyEvent")
        .field("id", Schema.STRING_SCHEMA)
        .field("payload", Schema.STRING_SCHEMA)
        .field("timestamp", Schema.INT64_SCHEMA)
        .build();

    @Override
    public void start(Map<String, String> props) {
        this.topic = props.get("topic");
        this.client = new LegacyClient(props.get("legacy.endpoint"));
    }

    @Override
    public List<SourceRecord> poll() throws InterruptedException {
        List<SourceRecord> records = new ArrayList<>();
        for (LegacyEvent event : client.fetchEvents()) {
            Struct value = new Struct(VALUE_SCHEMA)
                .put("id", event.getId())
                .put("payload", event.getPayload())
                .put("timestamp", event.getTimestamp());

            records.add(new SourceRecord(
                Collections.singletonMap("source", "legacy"),   // source partition
                Collections.singletonMap("offset", event.getOffset()), // source offset
                topic, null,                                     // topic, kafka partition
                Schema.STRING_SCHEMA, event.getId(),             // key schema + key
                VALUE_SCHEMA, value,                             // value schema + value
                event.getTimestamp()));
        }
        return records;
    }

    @Override
    public void stop() {
        if (client != null) client.close();
    }

    @Override
    public String version() {
        return "1.0.0";
    }
}

The source partition/offset map is what lets Connect resume from the last committed position after a restart, so it must uniquely identify your position in the upstream system. A common bug is making the source partition too coarse (e.g., a constant "source"): Connect tracks offsets per source partition, so collapsing everything into one partition means a single offset cursor and no way to parallelize or resume independent streams.

Managing Schemas Across the Development Lifecycle

Schemas have to move through dev, staging, and production, often with looser rules in dev and strict BACKWARD in prod. Manual export/import via the REST API does not scale, so register and compatibility-check schemas in CI alongside the application code that owns them.

The Schema Registry Maven plugin’s test-compatibility goal checks local schema files against a running registry and fails the build on a violation. The compatibility level used for the check is whatever the target subject is configured for in that registry — the goal does not take a compatibilityLevels parameter (that belongs to the separate set-compatibility goal):

<plugin>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-schema-registry-maven-plugin</artifactId>
    <version>7.6.0</version>
    <configuration>
        <schemaRegistryUrls>
            <param>http://schema-registry-staging:8081</param>
        </schemaRegistryUrls>
        <subjects>
            <orders-value>src/main/avro/orders.avsc</orders-value>
        </subjects>
    </configuration>
    <executions>
        <execution>
            <goals><goal>test-compatibility</goal></goals>
        </execution>
    </executions>
</plugin>

This catches incompatible changes before they reach production. Check against the registry whose rules you actually deploy under: if CI points at a dev registry set to NONE, the build passes and prod still rejects the schema. For promotion strategies and environment isolation see Managing Schemas Across Multiple Environments.

Monitoring and Operating Kafka Connect Clusters

Connect exposes metrics over JMX; scrape them with the JMX exporter into Prometheus. The MBeans you care about are kafka.connect:type=connector-task-metrics (per-task) and kafka.connect:type=connector-metrics (per-connector), plus the worker rebalance metrics under kafka.connect:type=connect-worker-rebalance-metrics.

There is an important gotcha here. The status attribute of connector-task-metrics reports running / paused / failed, but when a task actually fails its task-metrics MBean is no longer registered at all — the metric disappears rather than reporting failed (KAFKA-9066). An alert that looks for a JMX task metric equal to failed will therefore never fire. Detect failures off the connector-level status or, more reliably, by polling the REST status endpoint:

groups:
  - name: kafka_connect
    rules:
      # connector-metrics still reports status when a task dies; alert on FAILED.
      - alert: ConnectConnectorFailed
        expr: kafka_connect_connector_status{status="failed"} == 1
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Connector {{ $labels.connector }} is FAILED"
          description: "Connector has reported FAILED for more than 5 minutes."

Rising source lag (upstream position minus last committed offset) usually means too few tasks, network latency, or sink backpressure. Frequent rebalances point to worker instability — often GC pauses, undersized heaps, or a session.timeout.ms set too tight. (Incremental cooperative rebalancing, the default since Kafka 2.3, blunts but does not eliminate the impact of a flapping worker.) See Monitoring and Operating Kafka Connect Clusters and the Connect monitoring reference.

REST API Health Checks

Because the JMX failed-task signal is unreliable, the REST status endpoint is the source of truth for connector and task health. Build these into your runbooks:

# Worker liveness and version
curl -s http://connect-worker:8083/ | jq .

# Every connector with its task-level state in one call
curl -s "http://connect-worker:8083/connectors?expand=status&expand=info" | jq .

# Restart a single failed task (202 Accepted, asynchronous)
curl -X POST http://connect-worker:8083/connectors/mysql-inventory-connector/tasks/0/restart

# Restart the connector and only its failed tasks (KIP-745)
curl -X POST "http://connect-worker:8083/connectors/mysql-inventory-connector/restart?includeTasks=true&onlyFailed=true"

The restart connector endpoint’s includeTasks and onlyFailed query parameters (KIP-745, Kafka 3.0+) let you bounce only what is broken instead of restarting healthy tasks. The single-task restart (/tasks/0/restart) predates KIP-745 and is the right tool when exactly one task is wedged; reach for the connector-level restart when several tasks failed together (e.g., after a downstream outage).

Schema Validation and Serialization Formats

Pick the format by the constraint that actually binds you:

Schema Registry applies the same compatibility rules across all three, so the governance model is uniform regardless of format. See Using Protobuf with Kafka and Schema Registry and JSON Schema Support in Confluent Platform.

Configuring Protobuf Converters

key.converter=io.confluent.connect.protobuf.ProtobufConverter
value.converter=io.confluent.connect.protobuf.ProtobufConverter
key.converter.schema.registry.url=http://schema-registry:8081
value.converter.schema.registry.url=http://schema-registry:8081

JSON Schema is analogous with io.confluent.connect.json.JsonSchemaConverter. If you want the registry to reject invalid records rather than merely attach a schema, enable broker-side schema validation on the topic; consult the converter’s documentation for the validation-related properties in your Confluent Platform version, since the defaults attach a schema without strictly enforcing it.

Production Hardening and Capacity Planning

Run at least two Schema Registry instances behind a load balancer. They are stateless apart from the shared _schemas topic, so scaling out is cheap; one instance is elected primary for writes and the rest serve reads (and forward writes to the primary). The hard durability requirement is on _schemas itself — it must be compacted with a replication factor of 3. Never set a retention/delete policy on _schemas: deleting its records discards your entire schema history, after which existing data on the cluster becomes undecodable.

Connect needs more thought. Worker count, task distribution, and per-worker resources follow connector throughput and latency. A reasonable starting point is 2-4 CPU cores and a 4-8 GB heap per worker, then size from observed task throughput and rebalance behavior. Kafka Connect Architecture and Deployment Modes covers sizing.

Securing the Ecosystem

Schema Registry security builds on Kafka’s: TLS in transit, mTLS or SASL for authentication, and ACLs on _schemas (the registry’s principal needs read/write; everyone else needs read at most). Restrict the registry REST API to authorized clients — a registry exposed with no auth lets anyone change a subject’s compatibility level, which quietly disables the very enforcement you deployed it for.

For Connect, keep secrets out of connector JSON with a ConfigProvider (KIP-297). The example below uses the community HashiCorp Vault provider; note the real class name and vault.* parameters:

config.providers=vault
config.providers.vault.class=com.github.jcustenborder.kafka.config.vault.VaultConfigProvider
config.providers.vault.param.vault.address=http://vault:8200
config.providers.vault.param.vault.token=${file:/vault/token:VAULT_TOKEN}

Connector configs then reference secrets with the KIP-297 syntax ${provider:path:key} — for example ${vault:secret/mysql-debezium:password}, as used in the Debezium config above. The runtime resolves the placeholder before handing the config to the connector, so the secret never lands in the config topic in plaintext. One caveat: the REST API’s connector-config endpoint returns the resolved config to anyone who can read it, so the placeholder protects the config topic and disk, not necessarily an over-permissive REST endpoint — lock both down.

Backup and Disaster Recovery

Because all schema state is in _schemas, DR comes down to replicating that topic (plus Connect’s connect-configs, connect-offsets, connect-status) to the standby cluster. MirrorMaker 2 will do it, but watch the naming gotcha: the default DefaultReplicationPolicy renames replicated topics to <source-alias>.<topic>, so _schemas arrives as source._schemas and the standby registry — which only ever reads _schemas — sees nothing. Use IdentityReplicationPolicy so names are preserved:

# mm2.properties
clusters = source, dr
source.bootstrap.servers = kafka-source:9092
dr.bootstrap.servers = kafka-dr:9092

source->dr.enabled = true
source->dr.topics = _schemas, connect-configs, connect-offsets, connect-status

# Preserve topic names so the standby registry still finds _schemas
replication.policy.class = org.apache.kafka.connect.mirror.IdentityReplicationPolicy

replication.factor = 3
sync.topic.acls.enabled = true

IdentityReplicationPolicy is intended for one-way replication; do not use it for active-active flows, where same-named topics create a replication loop. There is a second, easily missed requirement: schema IDs are assigned by the source registry and embedded in every record, so the standby must serve the same ID-to-schema mapping. Replicating _schemas byte-for-byte preserves those IDs; rebuilding the standby registry from scratch would reassign IDs and make replicated data undecodable. After a failover the standby registry serves schemas from the replicated _schemas, and Connect resumes from the replicated offset topic.

Conclusion

Schema Registry and Kafka Connect are what make a Kafka deployment safe to grow: the registry turns “we agree on the data shape” into an enforced contract, and Connect turns one-off integrations into a managed, restartable fleet. The leverage is real, but so are the sharp edges — compatibility modes that dictate your rollout order, Debezium’s renamed config keys, JMX task metrics that vanish on failure, and MirrorMaker 2 renaming _schemas out from under the registry. Get those right and the patterns here — compatibility enforcement in CI, CDC with schema’d events, custom connectors that defer serialization to the converter, and ID-preserving DR replication — hold up under production load. The definitive references remain the Confluent Schema Registry documentation and the Apache Kafka Connect documentation.