Developing a Custom Source Connector for Legacy Systems

Legacy systems—mainframes, proprietary databases, fixed-width flat files, and aging ERP platforms—house critical enterprise data. Integrating them into an Apache Kafka event-driven architecture requires a purpose-built source connector when off-the-shelf options cannot handle custom wire formats, bespoke authentication, or idiosyncratic protocols. This guide provides a production-focused methodology for building that connector.

Prerequisites: Familiarity with the Kafka Connect Architecture and Deployment Modes and the Schema Registry & Connect Ecosystem.

The Source Connector Contract

A Kafka Connect source connector extends SourceConnector and produces SourceTask instances. The connector class handles configuration validation and splits work into per-task configurations; each task connects to the legacy system, extracts data in poll(), and returns SourceRecord objects. Connect’s framework producer writes those records to Kafka and persists their source offsets.

Production readiness demands more than functional correctness: connection pooling, backpressure awareness, graceful shutdown, and offset management that maps to the legacy system’s primitives (paginated API, batch file drop, terminal session, and so on).

public class LegacySourceConnector extends SourceConnector {
    private Map<String, String> configProperties;

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

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

    @Override
    public List<Map<String, String>> taskConfigs(int maxTasks) {
        // For a single logical source, one task config is typical.
        // Split by table, region, or partition key only if the system
        // supports concurrent, independent reads.
        List<Map<String, String>> configs = new ArrayList<>();
        configs.add(new HashMap<>(configProperties));
        return configs;
    }

    @Override
    public ConfigDef config() {
        return new ConfigDef()
            .define("legacy.host", ConfigDef.Type.STRING, ConfigDef.Importance.HIGH,
                    "Hostname of the legacy system")
            .define("legacy.port", ConfigDef.Type.INT, ConfigDef.NO_DEFAULT_VALUE,
                    ConfigDef.Range.between(1, 65535), ConfigDef.Importance.HIGH,
                    "Port of the legacy system")
            .define("legacy.batch.size", ConfigDef.Type.INT, 100,
                    ConfigDef.Range.atLeast(1), ConfigDef.Importance.MEDIUM,
                    "Number of records to fetch per poll");
    }

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

    @Override
    public void stop() {
        // Release connector-level resources.
    }
}

ConfigDef.Range.between(min, max) and ConfigDef.Range.atLeast(min) both accept Number, so integer literals work directly. NO_DEFAULT_VALUE makes legacy.port mandatory—there is no sensible default port for an arbitrary mainframe endpoint.

Offset Management for Legacy Sources

Offset management is the linchpin of resumable, at-least-once delivery. Modern databases offer transaction IDs or change-log positions; legacy systems rarely do. Your offset must be deterministic, resumable, and tolerant of duplicates.

Common patterns:

• Timestamp-based: Use a last-modified timestamp. Guard against clock skew and duplicate timestamps by combining the timestamp with a unique key (for example, a primary key).
• Pagination tokens: Persist the API cursor or page token.
• File sequence numbers: Use filename plus line number, or a content checksum.

Each SourceRecord carries a source-partition map and a source-offset map. Connect persists the offset only after the framework producer’s send for that record is acknowledged, then invokes SourceTask.commitRecord() per record (and commit() after a batch). Design for idempotency: if the worker crashes between the broker ack and the offset flush, records replay, so the connector—or downstream consumers—must tolerate or deduplicate them.

OffsetStorageReader.offset(partition) returns Map<String, Object> and returns null when no offset has been stored yet, so cast values explicitly and null-check the result:

@Override
public List<SourceRecord> poll() throws InterruptedException {
    List<SourceRecord> records = new ArrayList<>();
    Map<String, Object> partition = Collections.singletonMap("partition", "legacy-main");

    Map<String, Object> offset = context.offsetStorageReader().offset(partition);
    String lastOffset = (offset != null) ? (String) offset.get("position") : null;

    LegacyBatch batch = legacyClient.fetchSince(lastOffset, batchSize);
    if (batch.isEmpty()) {
        return null; // Block-then-return-null lets the task pause cleanly.
    }
    for (LegacyRecord rec : batch.getRecords()) {
        Map<String, Object> sourceOffset =
            Collections.singletonMap("position", rec.getCursor());
        records.add(new SourceRecord(
            partition,
            sourceOffset,
            topic,
            Schema.STRING_SCHEMA, rec.getKey(),
            Schema.STRING_SCHEMA, rec.getValue()
        ));
    }
    return records;
}

Returning null (rather than an empty list) when there is nothing to emit is the documented contract—it lets the task transition to PAUSED if requested. For systems with no cursor at all, use a snapshot-plus-incremental approach: bulk-load initially, record a high-water mark, then poll a change feed or re-read recent records. The Connector Development Guide covers the task lifecycle in detail.

Handling Legacy Protocols and Data Formats

Legacy systems speak protocols that predate REST and JSON: COBOL copybooks, fixed-width flat files, binary ASN.1, or proprietary TCP socket protocols. Your connector must translate these into structured Kafka messages.

Design principle: Encapsulate protocol handling in a dedicated client library within your connector project. This isolates parsing logic and makes it mockable. For binary protocols, use Java’s ByteBuffer or Netty. For mainframe copybook data, consider JRecord or cobol2json. Convert data into a canonical intermediate form (for example, an Avro GenericRecord) before building SourceRecord objects.

// Parse a fixed-width record and convert to an Avro GenericRecord.
ByteBuffer buffer = ByteBuffer.wrap(rawBytes);

byte[] idBytes = new byte[10];
buffer.get(idBytes); // ByteBuffer.get(byte[]) fills the array in place
String customerId = new String(idBytes, StandardCharsets.UTF_8).trim();

long balance = buffer.getLong();
int statusCode = buffer.getInt();

GenericRecord avroRecord = new GenericData.Record(schema);
avroRecord.put("customer_id", customerId);
avroRecord.put("balance", balance);
avroRecord.put("status_code", statusCode);

ByteBuffer.get(byte[]) reads exactly array.length bytes into the supplied array and advances the buffer position; it returns the buffer itself, not the bytes, so you read into a pre-sized array as shown. Mind the buffer’s byte order: mainframe data is typically big-endian, which is ByteBuffer’s default—call buffer.order(ByteOrder.LITTLE_ENDIAN) only if the source dictates it.

Register your Avro schema with the Schema Registry (programmatically or via the build plugin), then cache it at task startup to avoid a registry lookup per record.

Configuration, Packaging, and Deployment

Package the connector and its dependencies into a directory referenced by the worker’s plugin.path, or build a shaded uber-JAR placed in that path. (plugin.path is the supported isolation mechanism; avoid dumping connector JARs onto the worker classpath.) For production, run distributed mode—see Kafka Connect Architecture and Deployment Modes—for fault tolerance and rebalancing.

Never hardcode credentials. Use Connect’s ConfigProvider mechanism (KIP-297) to resolve sensitive values at runtime so secrets are never persisted in the config topic or exposed over the REST API. The bundled FileConfigProvider reads from a local properties file using ${file:/path:key} references; production deployments commonly back this with Vault or a cloud secrets manager.

Example connector configuration submitted via the REST API:

{
  "name": "legacy-mainframe-source",
  "config": {
    "connector.class": "com.example.LegacySourceConnector",
    "tasks.max": "1",
    "legacy.host": "mainframe.internal",
    "legacy.port": "2323",
    "legacy.batch.size": "500",
    "topic": "mainframe.customers",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",
    "config.providers": "file",
    "config.providers.file.class": "org.apache.kafka.common.config.provider.FileConfigProvider"
  }
}

Monitor health via GET /connectors/{name}/status and GET /connectors/{name}/tasks/{id}/status. Expose JMX metrics to Prometheus or Datadog. The source-task MBean is kafka.connect:type=source-task-metrics,connector=*,task=*; useful attributes include source-record-poll-rate (records polled per second), source-record-write-rate (records written to Kafka per second), and source-record-active-count (records polled but not yet fully written). See the Kafka Connect monitoring documentation for the full MBean catalog. Note that offset-commit-* metrics belong to sink-task metrics, not source tasks.

Testing and Failure Modes

Unit tests are necessary but not sufficient. Simulate legacy-system failures: slow responses, dropped connections, malformed data, and duplicate records. Run integration tests against a real broker and Schema Registry—Testcontainers spins up ephemeral instances—and mock the legacy endpoint with WireMock or a custom TCP proxy that injects latency and errors.

Handle these scenarios:

• Connection loss: Exponential backoff with jitter before retrying inside the task. Do not throw exceptions that permanently fail the task on transient errors; retry, and only surface a ConnectException once retries are exhausted.
• Schema evolution: Detect format changes. Fail fast on incompatible input and rely on Schema Registry compatibility checks to reject breaking changes at registration.
• Offset corruption: If a stored offset is invalid (for example, a cursor that has expired server-side), reset to the earliest available data and alert the operator rather than silently looping.

private void connectWithRetry() {
    int retries = 0;
    while (retries < maxRetries) {
        try {
            this.connection = legacyClient.connect();
            return;
        } catch (IOException e) {
            retries++;
            long waitTime = (long) (Math.pow(2, retries) * 1000)
                + ThreadLocalRandom.current().nextInt(1000);
            log.warn("Connection failed (attempt {}), retrying in {} ms", retries, waitTime);
            try {
                Thread.sleep(waitTime);
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
                throw new ConnectException("Interrupted during retry backoff", ie);
            }
        }
    }
    throw new ConnectException("Failed to connect after " + maxRetries + " retries");
}

A gotcha worth stating plainly: the dead-letter-queue mechanism does not apply to source connectors. The errors.tolerance setting and the converter/transformation error-reporting introduced by KIP-298 do run for source records, but errors.deadletterqueue.topic.name only routes failed records for sink connectors. For a source connector, set errors.tolerance to all to skip-and-log poison records (failures are written to the worker log, not to a DLQ topic), and build your own quarantine path—route bad records to a separate Kafka topic from inside poll() if you need to retain them. See the error-reporting section of the Connect documentation.

Performance Tuning and Capacity Planning

Legacy systems are often resource-constrained. Balance throughput against the source’s capacity rather than maximizing connector throughput in isolation.

Note that batch size, retry policy, and pool sizing are connector-specific configs you define; only worker-level settings such as offset.flush.interval.ms come from Connect itself. There is no built-in source-side poll.interval.ms—polling cadence is whatever your poll() implementation does (typically block until data is available, then return).

Constraint: Many legacy systems serialize access. If taskConfigs() cannot split the source into independent partitions, raising tasks.max does nothing—Connect will still run a single task—so optimize within that one task (larger batches, connection reuse, pipelined reads).

# Update the running connector's configuration via the REST API.
# PUT replaces the ENTIRE config, so every required field must be present.
curl -X PUT http://connect-worker:8083/connectors/legacy-mainframe-source/config \
  -H "Content-Type: application/json" \
  -d '{
    "connector.class": "com.example.LegacySourceConnector",
    "tasks.max": "1",
    "legacy.host": "mainframe.internal",
    "legacy.port": "2323",
    "legacy.batch.size": "200",
    "topic": "mainframe.customers",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }'

PUT /connectors/{name}/config replaces the connector’s configuration wholesale—omitting connector.class or the converters would break the connector—so always submit the complete config. Load-test in a staging environment that mirrors the legacy system’s performance profile, measure end-to-end latency from record creation to topic availability, and record it in the connector’s operational runbook because it feeds directly into downstream SLAs.

Conclusion

A custom source connector for legacy systems is a real engineering investment, but it unlocks decades of enterprise data for real-time processing. Adhere to the SourceConnector/SourceTask contract, design offset handling that is deterministic and replay-tolerant, parse legacy protocols in an isolated and testable client, and operationalize with the correct source-task metrics and realistic failure testing—remembering that DLQ routing is a sink-only feature. The result is a dependable bridge between legacy infrastructure and a modern event-driven platform.