This version is still in development and is not considered stable yet. For the latest stable version, please use Korvet 0.19!

Kafka Connect Integration

This guide covers running Kafka Connect against Korvet.

Overview

Kafka Connect is the standard framework for streaming data between Kafka and external systems (databases, object stores, message queues, search engines). Because Korvet implements the Kafka wire protocol, Connect workers treat it as a regular Kafka cluster: the framework — and with it the existing connector ecosystem (JDBC, S3, RabbitMQ, MQTT, Elasticsearch, and so on) — runs against Korvet with no Korvet-specific configuration.

Both worker deployment modes work:

  • Distributed mode — the mode production deployments use. The worker stores connector configuration, source offsets, and connector status in three compacted Korvet topics that it creates itself, and coordinates workers through Korvet’s group coordinator using the Connect rebalance protocol.

  • Standalone mode — a single worker with file-based offsets, useful for development and quick pipelines.

Connectors requiring exactly-once semantics are the exception: exactly-once source support (exactly.once.source.support) and transactional sink patterns rely on Kafka transactions, which Korvet does not implement. Run connectors with the default at-least-once delivery instead. See Kafka Compatibility.

Prerequisites

  • Korvet server running and reachable from the Connect worker on port 9092

  • A Kafka Connect distribution — the examples below use the apache/kafka Docker image, which ships the Connect runtime and the built-in FileStream connectors

Distributed Mode

A minimal connect-distributed.properties for a worker pointed at Korvet:

bootstrap.servers=<korvet-host>:9092
group.id=my-connect-cluster

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter

# Internal topics: the worker creates these in Korvet with cleanup.policy=compact
config.storage.topic=connect-configs
config.storage.replication.factor=1
offset.storage.topic=connect-offsets
offset.storage.replication.factor=1
offset.storage.partitions=25
status.storage.topic=connect-status
status.storage.replication.factor=1
status.storage.partitions=5

listeners=HTTP://0.0.0.0:8083
plugin.path=/opt/kafka/libs

Start the worker:

/opt/kafka/bin/connect-distributed.sh connect-distributed.properties

On startup the worker creates its three internal topics through the Kafka AdminClient with cleanup.policy=compact. Korvet honors the compaction config: a background pass keeps only the latest record per key, which is exactly what Connect relies on to rebuild connector state by replaying these topics. See Topic Management for details on log compaction.

Managing Connectors

Connectors are managed through the worker’s REST API, unchanged:

# Register a source connector
curl -X PUT -H "Content-Type: application/json" \
  --data '{
    "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "tasks.max": "1",
    "file": "/data/input.txt",
    "topic": "my-topic"
  }' \
  http://localhost:8083/connectors/file-source/config

# Check its status
curl http://localhost:8083/connectors/file-source/status

Standalone Mode

For a single-worker pipeline with file-based offsets, see the runnable example in samples/kafka-connect, which ships a Docker Compose stack wiring a standalone worker with a FileStream sink connector to Korvet.

Delivery Guarantees

Guarantee Status Notes

At-least-once

✅ Supported

The Connect default. Source offsets are committed to the compacted offset topic; sink progress is committed through consumer group offsets.

Exactly-once (source)

❌ Not supported

exactly.once.source.support requires Kafka transactions, which Korvet does not implement.

Troubleshooting

Worker Hangs Waiting for Internal Topics

If the worker logs errors creating or validating connect-configs/connect-offsets/connect-status, check that the configured topic names are not already taken by non-compacted topics. Connect refuses to run against internal topics whose cleanup.policy is not compact.

Worker Cannot Reach the Broker

When the worker runs in a container and Korvet on the host (or behind a load balancer), configure the advertised address so the bootstrap connection’s metadata points somewhere the worker can resolve:

export KORVET_BROKER_ADVERTISED_HOST=<reachable-hostname>
export KORVET_BROKER_ADVERTISED_PORT=9092

Next Steps