Kafka Connect

errors.log.enable=true logs each failed record's error context to the Connect worker log. errors.log.include.messages=true additionally logs the record's key/value contents. Including messages can leak sensitive payload data into logs.

You need bootstrap.servers (Kafka brokers), group.id (cluster identity), the three internal topic names with replication factors, key/value converters, plugin.path, and the REST listener. These let the worker connect to Kafka, join its cluster, and load plugins.

OffsetStorageReader lets a source task read back the last committed source offsets when it starts, so it can resume from where it left off. The task gets it from its SourceTaskContext and queries by source partition.

PUT /connectors/{name}/pause halts processing but keeps tasks assigned (state PAUSED). PUT /connectors/{name}/stop also halts but de-allocates the tasks (state STOPPED), freeing cluster resources. PUT /connectors/{name}/resume restarts a paused or stopped connector back to RUNNING.

tasks.max caps how many tasks a connector creates, which caps how many workers can share its load. Too low wastes workers (idle) and limits throughput; too high creates more tasks than the source can usefully split, adding overhead with no gain.

A sink connector subscribes to topics via the topics config (explicit list) or topics.regex (pattern). The tasks form a consumer group, so Kafka's group rebalance assigns each partition of those topics to exactly one task.

The Postgres connector uses logical replication: Postgres must run with wal_level=logical. A logical decoding output plugin (pgoutput, built in, or wal2json) turns WAL records into change events, delivered through a replication slot that tracks the last LSN the connector confirmed. The big risk: if the connector is down, the slot holds WAL on disk and can fill the disk.

The sink's converter usually doesn't match how the data was actually written. Identify the real on-topic format, align the sink's key/value converter and settings (e.g. schemas.enable, schema.registry.url) to it, and use error tolerance / a DLQ for poison records.

The framework tolerates errors in the parts it controls: converters (deserialization/serialization), SMTs, and the sink connector's put() handoff. It does NOT tolerate failures outside that boundary — like consumer/producer/offset-commit errors or framework/configuration errors, which still fail the task.

errors.retry.timeout is how long Connect keeps retrying a failed operation (default 0 = no retries; -1 = forever). errors.retry.delay.max.ms caps the backoff between attempts (default 60000). Retries run first; only after they're exhausted does errors.tolerance decide skip vs fail.

plugin.path is a list of directories where the worker finds connector/converter/transform JARs. Connect loads each plugin in an isolated classloader to avoid dependency conflicts. Misconfiguring it causes 'class not found' errors or dependency clashes.

Connect (KIP-618, Kafka 3.3+) can run source connectors exactly-once by writing the produced records and their source offsets together in a single Kafka transaction. You enable it cluster-wide with exactly.once.source.support=enabled, and the connector must declare exactly.once.support and define transaction boundaries.

Stop the connector first. For modern Connect (3.6+), use the REST offsets API: GET/DELETE/PATCH /connectors/{name}/offsets. DELETE wipes offsets so a source restarts from scratch; PATCH sets specific source offsets or sink consumer-group offsets. Older clusters edit offset.storage.topic (source) or use kafka-consumer-groups --reset-offsets (sink).

Use a ConfigProvider: put a placeholder like ${file:/path:key} or ${vault:...} in the connector config instead of the literal secret. Connect resolves it at runtime from the provider, so the secret never lives in the REST payload or the config topic, and GET /config shows the placeholder.

Connect runs each connector's Validator over the proposed config. PUT /connector-plugins/{class}/config/validate returns a structured per-field result (definitions, current values, recommended values, and per-field errors) without creating anything, so tooling can pre-flight a config and surface errors before deploying.

POST /connectors/{name}/restart restarts only the Connector instance by default. Add ?includeTasks=true to also restart tasks, and ?onlyFailed=true to limit the restart to FAILED instances. To restart just failed tasks: POST /connectors/{name}/restart?includeTasks=true&onlyFailed=true.

It's how long (default 5 minutes) the leader waits before reassigning a departed worker's tasks, hoping the worker returns and reclaims them. It avoids churn from quick restarts but keeps those tasks down during the wait if the worker is really dead.

Set listeners=https://host:port plus listeners.https.ssl.* keystore/truststore configs for TLS. For auth, plug in a REST extension via rest.extension.classes (e.g. BasicAuthSecurityRestExtension for HTTP Basic) since Connect has no built-in user store.

Implement org.apache.kafka.connect.transforms.Transformation<R extends ConnectRecord<R>>. Override apply(R) to return a transformed record, configure(Map) to read config, config() to declare a ConfigDef, and close() to release resources. Handle both schema and schemaless records, build a new record with record.newRecord(...), package it as a plugin, and put the Jar on the Connect plugin path.

SourceRecord describes data going INTO Kafka: target topic/partition, key, value, schemas, plus sourcePartition/sourceOffset that point back into the external system for resume. SinkRecord describes data coming OUT of Kafka: it carries the actual Kafka topic, kafkaPartition, and kafkaOffset of where it was consumed.

Debezium delivers at-least-once: after a crash it may re-emit some events because offsets are committed periodically, not per-event. Effective exactly-once is achieved downstream by idempotency — each event carries the primary key, so consumers/sinks upsert by key and re-applied duplicates are harmless. Kafka Connect 3.3+ added exactly-once source support that can make source-side delivery exactly-once.

Workers sharing a group.id use Kafka's group-membership protocol to elect a leader that assigns connectors and tasks. The original protocol stopped all work on every change (stop-the-world); incremental cooperative rebalancing (KIP-415) only reassigns the affected tasks, avoiding global pauses.

Use cooperative rebalancing (connect.protocol=compatible/sessioned) so only moving tasks pause. Set scheduled.rebalance.max.delay.ms above your per-worker restart time so restarting workers reclaim their own tasks. Restart one worker at a time and wait for the cluster to settle between each.

Set connector.client.config.override.policy=Principal on the worker, then give each connector its own credentials via producer.override./consumer.override.sasl.jaas.config (resolved from a ConfigProvider). Define per-principal broker ACLs so each tenant can only touch its own topics.

SMTs run in listed order and each consumes the previous one's output, so an SMT that renames or removes a field changes what later SMTs can see. You must order field-shaping before routing or extraction that depends on those fields, mind schema changes, predicate scoping, and tombstone handling.