Building Custom Kafka Connect Connectors

The built-in connector ecosystem covers most common integrations, but platform and SRE teams eventually hit a gap: a legacy database with no maintained connector, an internal API, or a proprietary wire format. A custom Kafka Connect connector lets you put that integration logic inside the Connect runtime, where it inherits the framework’s threading model, offset storage, configuration validation, and REST-based lifecycle management instead of reimplementing them. This page is a practitioner’s guide to designing, implementing, testing, and operating custom connectors within the broader Schema Registry & Connect Ecosystem, with a focus on production concerns: delivery semantics, schema evolution, error handling, and observability.

Before writing one, exhaust the cheaper options. Understanding where custom code fits requires familiarity with the Kafka Connect Architecture and Deployment Modes. The framework supplies the runtime; your connector supplies data-movement logic. Often an existing connector plus a chain of single message transforms (SMTs) gets you there. A custom connector is a long-lived maintenance liability, so justify it with a genuinely novel requirement, not a transform you could have configured.

Connector Development Fundamentals

Connectors come in two flavours: source connectors that ingest from external systems into Kafka topics, and sink connectors that write from Kafka topics to external systems. Both share a lifecycle managed by the runtime, but their task APIs differ. A source task partitions the external system, tracks per-partition offsets, and emits SourceRecord objects. A sink task receives SinkRecord collections and writes them to the destination, typically batching for efficiency.

The entry point is the abstract Connector class: extend SourceConnector for sources, SinkConnector for sinks. The connector class itself moves no data — it defines the configuration contract and, via taskConfigs(int maxTasks), decides how work is split into independent Task instances that the runtime distributes across workers. That split is the unit of parallelism, so taskConfigs should hand each task a non-overlapping slice of the source (e.g. a table range or a set of partitions), not maxTasks identical copies of the same config — duplicate slices mean duplicate data.

A minimal source connector skeleton:

public class CustomSourceConnector extends SourceConnector {
    private Map<String, String> configProps;

    @Override
    public void start(Map<String, String> props) {
        this.configProps = props;
    }

    @Override
    public Class<? extends Task> taskClass() {
        return CustomSourceTask.class;
    }

    @Override
    public List<Map<String, String>> taskConfigs(int maxTasks) {
        // Partition the source so each task owns a disjoint slice.
        // Here we split a list of logical partitions across maxTasks.
        List<List<String>> grouped = ConnectorUtils.groupPartitions(
            discoverSourcePartitions(), maxTasks);
        List<Map<String, String>> configs = new ArrayList<>();
        for (List<String> assignment : grouped) {
            Map<String, String> taskConfig = new HashMap<>(configProps);
            taskConfig.put("assigned.partitions", String.join(",", assignment));
            configs.add(taskConfig);
        }
        return configs;
    }

    @Override
    public ConfigDef config() {
        return CustomSourceConnectorConfig.CONFIG_DEF;
    }

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

    @Override
    public void stop() {
        // Release connector-level resources (e.g. a monitoring thread)
    }
}

org.apache.kafka.connect.util.ConnectorUtils.groupPartitions is the helper the built-in connectors use to fan a partition list across tasks; you do not have to use it, but you do have to ensure assignments are disjoint.

The task does the actual work. A SourceTask implements poll(), which returns a List<SourceRecord> (and returns null, not an empty list, when it has nothing to emit, so the runtime can honour pause requests). The framework calls poll() repeatedly; the task keeps its own state between calls. Each record carries a source partition and source offset map that the runtime persists to its internal offsets topic; on restart, SourceTaskContext.offsetStorageReader() lets you resume from the last committed position.

public class CustomSourceTask extends SourceTask {
    private CustomSourceConnectorConfig config;
    private final Map<String, Object> sourcePartition =
        Collections.singletonMap("source", "custom");
    private long currentOffset;

    @Override
    public void start(Map<String, String> props) {
        this.config = new CustomSourceConnectorConfig(props);
        // Resume from the last committed offset, if any.
        Map<String, Object> stored = context.offsetStorageReader()
            .offset(sourcePartition);
        this.currentOffset = stored == null
            ? 0L
            : (Long) stored.get("position");
        // Initialize connection to the external system
    }

    @Override
    public List<SourceRecord> poll() throws InterruptedException {
        List<ExternalRow> batch = fetchSince(currentOffset, config.getBatchSize());
        if (batch.isEmpty()) {
            return null; // nothing available; let the runtime pause if asked
        }
        List<SourceRecord> records = new ArrayList<>(batch.size());
        for (ExternalRow row : batch) {
            Map<String, Object> sourceOffset =
                Collections.singletonMap("position", row.position());
            records.add(new SourceRecord(
                sourcePartition,
                sourceOffset,
                config.getTopic(),
                null,                       // let the partitioner choose
                Schema.STRING_SCHEMA, row.key(),
                Schema.STRING_SCHEMA, row.value()));
            currentOffset = row.position();
        }
        return records;
    }

    @Override
    public void commit() throws InterruptedException {
        // Optional: only needed if the external system must also be told
        // which records Connect has durably persisted.
    }

    @Override
    public void stop() {
        // Close the external connection
    }

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

A note on delivery semantics, because it is widely misunderstood. By default, source connectors are at-least-once: the offsets topic is committed asynchronously, so a crash between producing a record and committing its offset replays that record. True exactly-once for source connectors arrives only with KIP-618, and it is off by default — you must set exactly.once.source.support=enabled on the worker (Kafka 3.3+), it requires distributed mode, and it is all-or-nothing across every source connector on the cluster. Even then, your poll()/offset logic must be deterministic for the guarantee to hold. Do not promise exactly-once in connector docs unless the worker is actually configured for it.

Configuration Management and Validation

ConfigDef declares configuration parameters with types, defaults, validators, importance, and documentation. Define the contract explicitly so the Connect REST API can validate a configuration (via PUT /connector-plugins/{name}/config/validate) before the connector starts, and so operators get precise error messages.

public class CustomSourceConnectorConfig extends AbstractConfig {
    public static final String TOPIC_CONFIG = "topic";
    private static final String TOPIC_DOC = "Kafka topic to write data to";

    public static final String BATCH_SIZE_CONFIG = "batch.size";
    private static final int BATCH_SIZE_DEFAULT = 100;
    private static final String BATCH_SIZE_DOC = "Records to fetch per poll";

    public static final String CONNECTION_URL_CONFIG = "connection.url";
    private static final String CONNECTION_URL_DOC = "Connection URL for the source system";

    public static final ConfigDef CONFIG_DEF = new ConfigDef()
        .define(TOPIC_CONFIG, Type.STRING, Importance.HIGH, TOPIC_DOC)
        .define(BATCH_SIZE_CONFIG, Type.INT, BATCH_SIZE_DEFAULT,
                Range.atLeast(1), Importance.MEDIUM, BATCH_SIZE_DOC)
        .define(CONNECTION_URL_CONFIG, Type.STRING, Importance.HIGH, CONNECTION_URL_DOC);

    public CustomSourceConnectorConfig(Map<String, String> props) {
        super(CONFIG_DEF, props);
    }

    public String getTopic()         { return getString(TOPIC_CONFIG); }
    public int getBatchSize()        { return getInt(BATCH_SIZE_CONFIG); }
    public String getConnectionUrl() { return getString(CONNECTION_URL_CONFIG); }
}

Validation can go beyond type checks. A custom ConfigDef.Validator (the ensureValid(String name, Object value) method) can enforce, for example, that a URL parses or a credential has the expected shape. ConfigDef.Recommender can suggest valid values dynamically — for instance, listing the tables visible to a JDBC source — which surfaces in the REST validation response and in UIs. For connectors that touch the Schema Registry & Connect Ecosystem, expose parameters for the registry URL, subject naming strategy, and compatibility, but note those are handled by the converter layer (see below), not by the connector itself.

Schema Integration and Evolution

A crucial separation: connectors produce records using Kafka Connect’s runtime-neutral data API (org.apache.kafka.connect.data.Schema and Struct), and the converter turns that into the wire format. Serialization format — Avro, Protobuf, JSON Schema — is a worker/connector configuration concern (key.converter, value.converter), not something the connector hard-codes. This is why the same connector can run with AvroConverter in one deployment and JsonConverter in another.

For structured data, build a Connect Schema with SchemaBuilder and populate a Struct; the converter handles registration with Schema Registry:

Schema valueSchema = SchemaBuilder.struct().name("com.example.Order")
    .field("id",     Schema.INT64_SCHEMA)
    .field("amount", Schema.FLOAT64_SCHEMA)
    .field("status", Schema.OPTIONAL_STRING_SCHEMA)
    .build();

Struct value = new Struct(valueSchema)
    .put("id", row.id())
    .put("amount", row.amount())
    .put("status", row.status());

SourceRecord record = new SourceRecord(
    sourcePartition, sourceOffset, topic,
    null,
    Schema.STRING_SCHEMA, row.key(),
    valueSchema, value);

With value.converter=io.confluent.connect.avro.AvroConverter and value.converter.schema.registry.url set, the converter registers (or looks up) the Avro schema derived from your Connect schema and embeds the schema ID in each message — far cheaper than shipping the full schema per record.

Sink connectors face the reverse: records arrive already deserialized into Connect Struct/Schema, and the task maps them to the destination model. When upstream topics evolve, the sink may see multiple schema versions; this is where Schema Evolution and Compatibility in Confluent Schema Registry matters. A resilient sink reads fields by name with defaults for optional/missing fields rather than assuming a fixed shape, and routes records it genuinely cannot handle to a dead-letter queue.

A sink configuration with Avro and a DLQ:

name=custom-sink-connector
connector.class=com.example.CustomSinkConnector
tasks.max=4
topics=source-topic
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://schema-registry:8081
value.converter.schema.registry.ssl.truststore.location=/etc/security/truststore.jks
errors.tolerance=all
errors.deadletterqueue.topic.name=dlq-custom-sink
errors.deadletterqueue.topic.replication.factor=3
errors.deadletterqueue.context.headers.enable=true

errors.deadletterqueue.topic.replication.factor defaults to 3; set it explicitly (e.g. 1) on a single-broker dev cluster or connector startup fails when it tries to create the DLQ topic.

Error Handling and Dead-Letter Queues

The framework’s record-processing error handling (KIP-298) is governed by two settings. errors.tolerance has exactly two values: none (the default — fail the task on the first error) and all (log and skip failed records). There is no deadletterqueue value for errors.tolerance; the DLQ is a separate feature enabled by setting errors.deadletterqueue.topic.name. When errors.tolerance=all and a DLQ topic is configured, the framework writes failed records there, and with errors.deadletterqueue.context.headers.enable=true it attaches headers describing the original topic/partition/offset and the exception.

Two important scope limits:

• The DLQ applies to sink connectors only — it captures failures in the converter, transformation, and put() stages. Source connectors have no DLQ; a source record that cannot be produced is handled by the producer’s own retries and the task’s error tolerance.
• errors.tolerance and the DLQ cover failures the framework sees (deserialization, SMTs, the sink’s write path under errors.* retry handling). They do not cover exceptions you throw and catch entirely inside your own poll() loop — that logic is yours to write.

Framework-level retries for retriable failures are bounded by errors.retry.timeout (total retry duration in ms; 0 disables, -1 is infinite) and errors.retry.delay.max.ms (the cap on exponential backoff, default 60000).

For a source connector reaching out to an external system, you implement your own retry inside poll():

private List<SourceRecord> pollWithRetry() throws InterruptedException {
    int maxRetries = 3;
    long baseBackoffMs = 1000;
    for (int attempt = 0; attempt < maxRetries; attempt++) {
        try {
            return fetchRecords();
        } catch (TransientException e) {
            if (attempt == maxRetries - 1) {
                // Surface as retriable so the runtime restarts the task
                // after a backoff instead of killing it permanently.
                throw new RetriableException("Exhausted in-poll retries", e);
            }
            long backoff = baseBackoffMs * (1L << attempt); // 1s, 2s, 4s
            log.warn("Transient failure on attempt {}, retrying in {}ms",
                     attempt + 1, backoff);
            Thread.sleep(backoff);
        }
    }
    return null;
}

Distinguish transient from permanent failures. Throw org.apache.kafka.connect.errors.RetriableException for conditions the runtime should retry (it backs the task off and retries rather than marking it FAILED). Throw ConnectException (or let other exceptions propagate) for permanent faults — bad credentials, an unrecoverable schema mismatch — so the task fails fast and alerts operators instead of looping forever.

Testing Custom Connectors

Test in layers: unit tests for configuration and record-mapping logic, integration tests against a real (embedded) Kafka, and end-to-end tests against the actual external system in staging.

Unit-test the Connector and Task classes with external dependencies mocked. ConfigDef parsing deserves explicit coverage — missing or malformed config is a leading cause of production incidents:

@Test
public void resolvesProvidedConfig() {
    Map<String, String> props = Map.of(
        "topic", "test-topic",
        "connection.url", "https://example.com/api");
    CustomSourceConnectorConfig config = new CustomSourceConnectorConfig(props);
    assertEquals("test-topic", config.getTopic());
    assertEquals("https://example.com/api", config.getConnectionUrl());
}

@Test
public void rejectsMissingRequiredConfig() {
    Map<String, String> props = Map.of("topic", "test-topic"); // no connection.url
    assertThrows(ConfigException.class,
        () -> new CustomSourceConnectorConfig(props));
}

For integration tests, Apache Kafka ships org.apache.kafka.connect.util.clusters.EmbeddedConnectCluster, a test-only helper in the connect-runtime test-jar that boots an in-process Kafka (KRaft) plus Connect cluster and exposes the REST API. A typical test starts the embedded cluster, PUTs the connector config, produces or expects records, and asserts on delivery and offset progress. (This class is for testing only and is not a supported production embedding API; there is no equivalent embedded Schema Registry in Apache Kafka itself — use Testcontainers or the Confluent test utilities if you need one.) The JDBC and MirrorMaker 2 connectors in the Kafka and Confluent source trees are good worked examples.

End-to-end tests belong in a staging environment that mirrors production. Exercise the failure modes that actually bite connectors: the external system going slow or unavailable, network partitions, and schema changes mid-stream. Deliberately injecting these (chaos testing) is how you discover whether your timeout, backoff, and DLQ settings are right before an incident does it for you.

Operational Considerations and Monitoring

Connect exposes metrics over JMX under the kafka.connect domain. Use the real MBeans and attributes — these are the ones in the Kafka Connect monitoring docs:

• kafka.connect:type=connect-worker-metrics,connector="{connector}" → connector-total-task-count, connector-running-task-count, connector-failed-task-count, connector-paused-task-count. A non-zero connector-failed-task-count is your page-worthy signal, and running should equal total under healthy operation.
• kafka.connect:type=source-task-metrics,connector="{connector}",task="{task}" → source-record-poll-rate, source-record-write-rate, source-record-active-count, poll-batch-avg-time-ms.
• kafka.connect:type=sink-task-metrics,connector="{connector}",task="{task}" → sink-record-read-rate, sink-record-send-rate, offset-commit-completion-rate, offset-commit-skip-rate, put-batch-avg-time-ms. Note that offset-commit metrics live here, on the sink task — not on source tasks.
• kafka.connect:type=task-error-metrics,connector="{connector}",task="{task}" → total-record-errors, total-records-skipped, deadletterqueue-produce-requests. Watch these to know your DLQ and error tolerance are actually doing something.
• kafka.connect:type=connector-metrics,connector="{connector}" → status (e.g. running, failed, paused), plus connector-class/connector-type/connector-version.

There is no connector-task-count metric and no source-side offset-commit-completion-rate; ignore any dashboard that references them.

Application-specific metrics (API latency, upstream error rate, rate-limit headroom) are valuable, but be aware the public Connect API does not give connector code a handle to the runtime’s metrics registry — SourceTaskContext/SinkTaskContext expose configs and the offset reader, not a metrics() method. To publish your own metrics, register your own MBeans with the platform MBeanServer, or emit via a standard library (Micrometer, Dropwizard, or an OpenTelemetry agent) the same way any JVM service would. Keep the names in your own MBean domain so they do not collide with kafka.connect.

Logging uses SLF4J. Log lifecycle events and config summaries at INFO, record-level detail at DEBUG (off in production), and operator-actionable failures at ERROR. Consistent field names make aggregation and alerting tractable.

The Kafka Connect REST API drives runtime operations:

# Connector and per-task status
curl -s http://connect-worker:8083/connectors/custom-source/status | jq .

# Restart a single failed task (add ?onlyFailed=true on the connector
# endpoint to restart only failed tasks in one call)
curl -X POST http://connect-worker:8083/connectors/custom-source/tasks/0/restart

# Current connector configuration
curl -s http://connect-worker:8083/connectors/custom-source/config | jq .

Manage resources carefully. A connector holding persistent connections must pool them and release everything in stop() on both the connector and task — stop() runs on every rebalance and restart, and leaked sockets or threads accumulate fast. Sinks that buffer large batches need backpressure (bounded buffers, blocking on put()) so a slow destination produces lag rather than OutOfMemoryError.

Deployment follows the plugin model. Place the connector and its dependencies in a directory listed in the worker’s plugin.path; Connect (KIP-146) loads each plugin in an isolated classloader, so two connectors can depend on conflicting library versions without clashing. Counter-intuitively, the directory-of-JARs layout is the recommended one — an uber-JAR built with Maven Shade or Gradle Shadow also works, but shading does nothing for cross-plugin isolation (that comes from plugin.path) and can bundle classes that conflict with the runtime. Internal connectors typically ship via CI/CD that builds, tests, and stages the plugin directory to each worker; Confluent Hub is the public distribution channel.

When producing structured data, mind the subject naming strategy. The default TopicNameStrategy derives the subject from the topic; switch to RecordNameStrategy or TopicRecordNameStrategy when one topic carries multiple record types, or schema registration fails and surfaces as task failures. This too is a converter setting (value.converter.value.subject.name.strategy), not connector code.

Finally, plan upgrades. Connect supports rolling restarts of a plugin when the connector class name and configuration stay compatible. Breaking changes — a renamed connector class or an incompatible config schema — mean standing up a new connector and migrating, which has to be coordinated with topic retention and downstream consumers. Keeping configuration and output formats backward-compatible is what lets you roll out new versions gradually instead of with a flag day.