Kafka Connect

Kafka Connect Exists Because Integration Code Is Repetitive and Dangerous to Hand-Write

Every team that adopts Kafka eventually needs to move data between Kafka and something that is not Kafka — a Postgres database, an Elasticsearch cluster, an S3 bucket, a Salesforce API. The naive approach is to write a small service: connect to the database, poll for new rows, produce them to a topic. Write a second small service: consume from a topic, batch the records, write them to S3. Multiply this by every system you need to integrate with, and you end up with a fleet of one-off scripts, each with its own bugs around offset handling, retries, backoff, schema drift, and restart behavior.

Kafka Connect is a framework, shipped as part of Kafka itself, purpose-built to eliminate that repetition. It is not a new kind of broker and it is not a stream processing engine. It is a standardized runtime for running connectors — pluggable, configuration-driven components that move data between Kafka and external systems, without you writing a bespoke producer or consumer application for each integration.

The trade-off is that Connect is only as good as the connector plugin available for your system. For common integrations — relational databases, S3, Elasticsearch, MongoDB, Salesforce, Snowflake — mature, battle-tested connectors already exist, usually from Confluent, the community, or the vendor itself. For a bespoke internal system with business logic that does not map cleanly onto "read a record, write a record," you are often better off with a hand-written producer or consumer, which this module covers in the last section.

Source Connectors Pull Data In, Sink Connectors Push Data Out

Connect has exactly two categories of connector, and the direction of data flow is the entire distinction. Every connector you will ever configure is one or the other.

Source connectors — external system into Kafka

A source connector reads data from an external system and produces it into one or more Kafka topics. It acts as the producer so you do not have to write one. Common source connectors include a JDBC source connector that polls a relational database's tables and turns new or changed rows into Kafka records, and Debezium — the dominant open-source project for change data capture (CDC) — which reads a database's write-ahead log or binlog directly (rather than polling tables) and emits a Kafka record for every insert, update, and delete, including the before-and-after row state.

Sink connectors — Kafka into an external system

A sink connector consumes records from one or more Kafka topics and writes them to an external system. It acts as the consumer so you do not have to write one. Common sink connectors include an S3 sink connector that batches records and writes them as partitioned Parquet or JSON files in an S3 bucket, an Elasticsearch sink connector that indexes records for search, and a JDBC sink connector that upserts records into a relational database table.

   External system                 Kafka                External system
  ┌────────────────┐    SOURCE    ┌───────┐    SINK    ┌────────────────┐
  │  Postgres (CDC) │ ───────────▶│ topic  │───────────▶│  S3 bucket      │
  │  via Debezium   │  connector   │  logs  │  connector │  (Parquet)      │
  └────────────────┘              └───────┘             └────────────────┘

  Source connector = Connect acting as a PRODUCER on your behalf.
  Sink connector   = Connect acting as a CONSUMER on your behalf.
  A single Connect cluster commonly runs both kinds of connector
  at once, for completely unrelated pipelines.

Standalone Mode Is for Local Testing. Distributed Mode Is for Production.

Connect workers — the JVM processes that actually run connectors — can run in one of two modes. The mode is a deployment decision, not a per-connector setting; a given Connect cluster is entirely standalone or entirely distributed.

Standalone mode

A single worker process runs all configured connectors and tasks. Configuration lives in a local properties file on disk, and offsets are stored in a local file as well. There is no fault tolerance — if the worker process dies, every connector it was running stops, and nothing reassigns the work elsewhere. Standalone mode is appropriate for local development, quick experiments, and single-machine edge deployments where a Connect cluster would be overkill (for example, shipping logs from one specific edge device).

Distributed mode

Multiple worker processes form a Connect cluster. Connector configurations are submitted through a REST API rather than a local file, and are stored in Kafka itself (an internal config topic). Worker processes coordinate through Kafka's group membership protocol — the same underlying mechanism consumer groups use — to divide the work of running connectors and their tasks across available workers. If a worker fails, the connectors and tasks it was running are automatically rebalanced onto the surviving workers. This is the mode every production Kafka Connect deployment runs in.

The rest of this module assumes distributed mode, because that is what you will actually operate in production, and because the REST API workflow it uses is also the more instructive one to learn — you interact with Connect the same way whether you manage two connectors or two hundred.

A Worked Example — Submitting a JDBC Source Connector via the REST API

In distributed mode, every Connect worker exposes a REST API, typically on port 8083. You do not SSH into a machine and edit a file to add, update, or remove a connector — you send an HTTP request. This is deliberate: it means connector management can be automated, version-controlled, and driven from CI/CD the same way any other infrastructure configuration is.

Consider a concrete case: FreshCart, a grocery delivery company, wants every new row inserted into its Postgres orders table to appear as an event on a Kafka topic, so the fulfillment service can react to it without polling the database directly. Here is the connector configuration, submitted as JSON.

{
  "name": "freshcart-orders-jdbc-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "2",
    "connection.url": "jdbc:postgresql://db.internal:5432/freshcart",
    "connection.user": "connect_reader",
    "connection.password": "${file:/secrets/connect-creds.properties:db_password}",
    "table.whitelist": "orders",
    "mode": "incrementing",
    "incrementing.column.name": "order_id",
    "topic.prefix": "freshcart.pg.",
    "poll.interval.ms": "5000",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }
}

• ✓connector.class — the fully-qualified Java class of the connector plugin. This is how Connect knows which plugin JAR to load.
• ✓tasks.max — the upper bound on how many parallel tasks this connector can split its work into (see Part 04b below on tasks vs connectors).
• ✓connection.url / connection.user / connection.password — connector-specific properties; the JDBC source connector defines these, an S3 sink connector would define completely different ones.
• ✓mode: "incrementing" — tells the JDBC source connector how to detect new rows: by watching a strictly increasing column (order_id here) rather than a timestamp column or full-table dumps.
• ✓topic.prefix — new rows from the orders table land on the topic freshcart.pg.orders (prefix + table name).
• ✓key.converter / value.converter — covered in depth in Part 07; these control how record keys and values are serialized onto the topic.

Submitting this configuration is one HTTP POST to the cluster's REST endpoint.

curl -X POST http://connect-worker-1:8083/connectors \
  -H "Content-Type: application/json" \
  -d @jdbc-source-orders.json

{
  "name": "freshcart-orders-jdbc-source",
  "config": { ... },
  "tasks": [
    { "connector": "freshcart-orders-jdbc-source", "task": 0 },
    { "connector": "freshcart-orders-jdbc-source", "task": 1 }
  ],
  "type": "source"
}

Connectors vs tasks — where the parallelism actually lives

A connector itself does not move any data. It is a coordinator: it splits the overall job into one or more tasks, up to the tasks.max limit, and it is the tasks that actually read from or write to the external system, running on Connect workers across the cluster. For the JDBC source connector above, with tasks.max: 2 and only one table in the whitelist, Connect will typically run just one task, since there is only one table to split work across. A source connector reading five tables with tasks.max: 5 could run five tasks in parallel, each responsible for one table, spread across however many workers the cluster has.

Connect Tracks Its Own Offsets — Separate from Consumer Group Offsets

A common point of confusion: Connect offsets are not the same thing as Kafka consumer group offsets, even though both are called "offsets" and both serve the same conceptual purpose — remembering how far progress has gotten so work is not repeated or skipped after a restart.

Source connector offsets — position in the external system

A source connector's job is to track its position within the external system it is reading from, not within Kafka. For the JDBC source connector in Part 04, the offset is the lastorder_id value it has successfully read and produced. For Debezium reading a database's write-ahead log, the offset is a log sequence number (LSN) or binlog file-and-position pair. These offsets are stored by Connect itself, in an internal Kafka topic (by default namedconnect-offsets in distributed mode), completely independent of any consumer group.

A source connector does not consume from a Kafka topic at all --
it reads from Postgres, MongoDB, a filesystem, an API. There is no
Kafka partition and offset to track on the READ side.

Instead, Connect stores a (source partition -> source offset) pair
per connector, where "source partition" is a connector-defined
concept -- for the JDBC connector, one source partition per table:

  source partition: {"table": "orders"}
  source offset:     {"incrementing": 48213}

On restart, the JDBC source connector reads this back from the
connect-offsets topic and resumes with:
  "SELECT * FROM orders WHERE order_id > 48213 ORDER BY order_id"

Sink connector offsets — ordinary consumer group offsets

A sink connector, by contrast, genuinely is a Kafka consumer under the hood — it consumes records from a topic. Its progress is tracked exactly the way any consumer group's progress is tracked: through the standard __consumer_offsets internal topic, using a consumer group ID derived from the connector name. This is the one place where Connect's offset model maps directly onto ordinary Kafka consumer semantics.

Single Message Transforms — Lightweight In-Flight Record Editing

Often you need a small, mechanical change to every record passing through a connector — rename a field, drop a sensitive column, add a static field, route records to different topics based on a value — without writing and deploying a full transformation pipeline. Single Message Transforms (SMTs) are Connect's answer: small, chainable, configuration-only transformations applied to each record as it passes through a connector, before a source connector's record reaches Kafka, or before a sink connector's record reaches the external system.

A worked example — masking a sensitive field and renaming a column

Continuing the FreshCart example: the orders table includes acustomer_email column, which should not be broadcast onto a Kafka topic that many internal services can read, and a legacy column name, cust_id, that downstream consumers expect renamed to customer_id.

{
  "transforms": "maskEmail,renameCustomerId",

  "transforms.maskEmail.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.maskEmail.fields": "customer_email",
  "transforms.maskEmail.replacement": "REDACTED",

  "transforms.renameCustomerId.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
  "transforms.renameCustomerId.renames": "cust_id:customer_id"
}

The transforms property lists the transform names in the order they are applied — order matters, because each transform receives the output of the one before it. Each named transform then gets its own transforms.<name>.type andtransforms.<name>.* configuration properties, exactly like a connector gets its own connector.class and connector-specific properties.

Converters — How Connect Serializes Record Keys and Values

A converter is the component that translates between Connect's internal, in-memory representation of a record and the actual bytes stored on a Kafka topic (for a source connector) or read from a Kafka topic (for a sink connector). Every connector configuration sets akey.converter and a value.converter, and it is entirely normal — and common — for the two to differ, since keys are often simple strings or numbers while values carry the full record structure.

JSON converter — simple, but no schema enforcement

org.apache.kafka.connect.json.JsonConverter serializes records as plain JSON. It is simple to inspect and debug — you can read the raw bytes on a topic with any JSON-aware tool — but it carries no schema enforcement at all. Nothing prevents a producer from silently changing a field's type or removing a field, and every consumer discovers this only when its own deserialization or business logic breaks.

Avro converter with Schema Registry — schema enforcement across the pipeline

io.confluent.connect.avro.AvroConverter serializes records as Avro binary, registering the record's schema with a separate Schema Registry service the first time it is used, and storing only a small schema ID alongside the compact binary payload in each Kafka record after that. Every consumer of the topic — whether another Connect sink connector, a Kafka Streams application, or a hand-written consumer — looks up that schema ID against the same Schema Registry to know exactly how to deserialize the bytes, and the registry can be configured to reject schema changes that would break existing consumers.

# With JsonConverter -- readable, unenforced:
{"order_id": 48213, "customer_id": "C-991", "total_cents": 4599}

# With AvroConverter + Schema Registry -- compact binary, schema-checked:
# (not human-readable on the wire -- shown here as what it decodes to)
magic byte | schema_id=117 | <avro-encoded binary payload>

# A downstream consumer with AvroConverter configured:
# 1. Reads schema_id=117 from the record header
# 2. Fetches schema 117 from the Schema Registry (cached after first lookup)
# 3. Decodes the binary payload using that exact schema
# If the producer's schema changes incompatibly, the Schema Registry
# rejects the new schema at registration time -- before it ever
# reaches a topic and breaks a downstream consumer.

What Delivery Guarantee Does Connect Actually Give You?

Because a source connector is really a producer and a sink connector is really a consumer, the delivery guarantees from earlier modules apply directly, and it is worth being precise about which guarantee you are actually getting by default.

Source connectors — at-least-once by default

A source connector commits its source offset only after the corresponding record has been successfully produced to Kafka. If the connector crashes after producing a record but before persisting the updated offset, it will re-read and re-produce that same record on restart — at-least-once delivery, meaning downstream consumers can see the same record twice. Some source connectors, including recent versions of Debezium with idempotent producer settings enabled and exactly-once support in newer Connect versions, can reduce this to effectively-once for specific setups, but the safe default assumption for any source connector is at-least-once.

Sink connectors — depends on the target system's own idempotency

A sink connector, being a consumer, commits its Kafka offset after writing to the external system. The same at-least-once risk applies in the other direction: a crash between writing to the external system and committing the offset causes the same record to be written again on restart. Whether that matters depends entirely on whether the write to the external system is naturally idempotent — an upsert into a database keyed by a unique ID tolerates a duplicate write with no visible effect, while an S3 sink appending records to a file, or a message queue that does not deduplicate, can end up with genuine duplicates.

Errors Tolerance and Dead Letter Queues — Connect Has Its Own Built-In DLQ Pattern

A hand-written consumer needs custom code to route a poison message to a dead letter queue rather than blocking the whole partition, as covered in the message-brokers module. Connect builds this pattern in directly, as a small set of connector-level configuration properties, so you rarely need to write DLQ-handling logic yourself for a sink connector's conversion or transformation failures.

errors.tolerance — stop on the first bad record, or skip it and keep going

By default, errors.tolerance is none — the first record that fails to convert, transform, or be written to the target system fails the entire connector task, and the task stops until an operator intervenes. Setting errors.tolerance toall tells the connector to skip the offending record, log the failure, and keep processing subsequent records — the Connect equivalent of the retry-then-DLQ loop covered for hand-written consumers.

{
  "errors.tolerance": "all",
  "errors.log.enable": "true",
  "errors.log.include.messages": "true",

  "errors.deadletterqueue.topic.name": "freshcart.orders.dlq",
  "errors.deadletterqueue.topic.replication.factor": "3",
  "errors.deadletterqueue.context.headers.enable": "true"
}

With these properties set, a record that fails deserialization, an SMT, or the write to the target system is routed to the configured DLQ topic instead of stopping the task, and — witherrors.deadletterqueue.context.headers.enable set — Connect attaches headers to the DLQ record recording exactly which connector, task, and stage the failure happened at, and the original topic, partition, and offset, mirroring the manually-constructed DLQ event structure covered in the message-brokers module, but produced automatically by the framework.

Monitoring and Operating a Connect Cluster in Production

Connect exposes its own set of JMX metrics, separate from broker and consumer metrics, and because a single Connect cluster commonly runs many unrelated connectors at once, monitoring needs to be per-connector and per-task, not just cluster-wide.

Operationally, the single highest-leverage practice is alerting on any task transitioning to FAILED state, since Connect does not automatically retry a permanently failed task — it sits idle until an operator issues a POST /connectors/{name}/tasks/{taskId}/restart or diagnoses and fixes the underlying cause. A connector silently stuck in FAILED for hours is one of the most common causes of a "why hasn't data shown up in three hours" incident, and it produces no error on the producing or consuming side of the pipeline — from outside Connect, data simply stops flowing.

# Pseudocode for a monitoring check run every minute per connector:

for connector in list_connectors():
    status = get_connector_status(connector)
    if status.connector.state == "FAILED":
        alert(f"Connector {connector} itself has FAILED")
    for task in status.tasks:
        if task.state == "FAILED":
            alert(f"Connector {connector} task {task.id} has FAILED: "
                  f"{task.trace}")
        if task.state == "PAUSED" and not connector.expected_paused:
            alert(f"Connector {connector} task {task.id} is unexpectedly PAUSED")

How a Distributed Connect Cluster Actually Rebalances Work

A distributed Connect cluster's workers coordinate the same way a consumer group does, because under the hood they use the same group membership protocol described in Module 03 — one worker acts as the group leader, and the set of connectors and tasks is divided among all currently active workers. Adding a worker, removing one, or a worker crashing all trigger a rebalance, in which tasks are redistributed across whichever workers are currently part of the cluster.

Incremental cooperative rebalancing — the current default

Older Connect versions used eager rebalancing: every worker stopped every task it was running the moment any rebalance began, even tasks that would be reassigned right back to the same worker, and the whole cluster paused until every worker had its new assignment. Current Connect versions default to incremental cooperative rebalancing — the same underlying idea as the cooperative consumer group protocol covered in Module 03 — where only the specific tasks that actually need to move are stopped and reassigned, and every other task on every unaffected worker keeps running uninterrupted through the rebalance.

Before: worker-1 runs tasks [A-0, A-1, B-0]
        worker-2 runs tasks [B-1, C-0]

worker-3 joins the cluster.

Eager rebalancing (old):
  ALL five tasks are revoked across worker-1 and worker-2.
  Cluster is idle while a brand new full assignment is computed
  and handed out -- even A-0, which ends up back on worker-1.

Incremental cooperative rebalancing (current default):
  Only the minimal set of tasks needed to balance load moves --
  e.g. B-0 moves from worker-1 to worker-3, C-0 moves from
  worker-2 to worker-3. A-0, A-1, and B-1 never stop running.
  Net result: 2 tasks paused briefly, not all 5.

The practical takeaway for operating a Connect cluster is that scaling it out — adding workers to handle more connectors or higher-throughput tasks — is a low-disruption operation on current Connect versions, similar to how a rolling deploy of a well-configured consumer group is low disruption. This is part of why distributed mode is unambiguously the right choice for any production deployment, beyond just the fault-tolerance argument from Part 03.

Header Converters, and Why Topic and Connector Naming Discipline Matters at Scale

Kafka records carry an optional set of key-value headers separate from the record's key and value — small pieces of metadata attached to a record without needing to be part of its actual payload schema. Connect has a third converter setting, header.converter, alongsidekey.converter and value.converter, controlling how those headers are serialized. Most teams leave this at the simple default,org.apache.kafka.connect.storage.SimpleHeaderConverter, since headers are typically small, loosely-typed metadata (a trace ID, a source system tag) rather than data that benefits from the same schema enforcement the record value gets from Avro and Schema Registry.

Naming conventions become an operational necessity, not a nicety

A single Connect cluster in a mid-sized company commonly runs dozens to low hundreds of connectors within a year of adoption — one per table being CDC'd, one per destination system, often multiplied across environments. Without a naming convention, `GET /connectors` returns an unreadable list, and figuring out which connector produces to which topic, and why a given topic suddenly has no new data, becomes a manual archaeology exercise.

Connector name pattern:
  {source-or-sink}-{system}-{entity}-{env}

  source-postgres-orders-prod
  source-postgres-customers-prod
  sink-s3-orders-archive-prod
  sink-elasticsearch-orders-search-prod

Topic name pattern (source connectors):
  {company}.{source-system}.{entity}

  freshcart.pg.orders
  freshcart.pg.customers

DLQ topic pattern (Part 09):
  {original-topic}.dlq

  freshcart.pg.orders.dlq

The specific convention matters less than having one applied consistently from the very first connector — retrofitting a naming scheme across dozens of already-deployed connectors and their downstream consumers is a much larger, riskier migration than establishing the pattern before the second connector is ever created.

Exactly-Once Support for Source Connectors — A Newer, Connector-Specific Guarantee

Part 08 established that source connectors default to at-least-once delivery. Newer versions of Connect (Kafka 3.3 and later) support an opt-in exactly-once mode for source connectors,exactly.once.source.support, but — unlike Kafka Streams' exactly_once_v2from Module 14, which applies uniformly to any Kafka Streams topology — exactly-once source support in Connect must be explicitly implemented by each individual connector plugin. Enabling the worker-level setting does nothing on its own for a connector plugin that was never built to support it.

# On every worker's connect-distributed.properties:
exactly.once.source.support=enabled

# On the specific connector's config, IF that connector plugin
# actually implements transactional source support:
{
  "name": "freshcart-orders-jdbc-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "transaction.boundary": "poll",
    ...
  }
}

# A connector plugin that does NOT implement this support simply
# ignores the setting and continues operating at-least-once --
# Connect does not error out, which makes this easy to misconfigure
# and assume a guarantee that was never actually granted.

In practice, most teams that need a true end-to-end exactly-once guarantee for a database integration reach for Debezium specifically, since it has invested heavily in transactional source support and is the most widely deployed CDC connector in production Kafka deployments — but even then, the guarantee only covers the Debezium-to-Kafka leg. Anything downstream still needs its own idempotency handling, exactly as covered for sink connectors in Part 08, unless that downstream consumer is itself a Kafka Streams application running exactly_once_v2 against the same cluster.

The pragmatic default worth remembering across this entire module: assume at-least-once unless you have specifically verified otherwise for both the connector plugin in use and the version of Connect it is deployed on, since the exactly-once story for source connectors has changed across Kafka versions and is not uniformly available the way it is for Kafka Streams topologies covered in Module 14.

That framing — check the specific plugin and version rather than trusting a framework-wide assumption — is also the right instinct for converter compatibility (Part 07), SMT behavior across Connect versions (Part 06), and error-handling configuration defaults (Part 09): Connect standardizes the operational surface across every connector, but the actual guarantees underneath that surface are still connector-specific, and reading the specific plugin's documentation before depending on any of them in production is not optional diligence — it is the only way to know what you actually have.

Deploying Connect Workers — What Actually Runs in Production

A Connect worker is a JVM process started from the Kafka distribution'sconnect-distributed.sh script (or the equivalent entry point in a container image), pointed at a properties file containing cluster-wide settings — bootstrap servers, the internal config/offset/status topic names, and converter defaults that individual connectors can override. In production, teams almost always run Connect workers as containers under Kubernetes or a similar orchestrator, precisely because the cooperative rebalancing behavior from Part 10b tolerates workers being added, removed, or replaced gracefully — a property container orchestration takes advantage of naturally during rolling deploys and autoscaling.

bootstrap.servers=broker-1:9092,broker-2:9092,broker-3:9092
group.id=freshcart-connect-cluster

# Internal topics Connect creates and manages itself
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

# Cluster-wide default converters -- individual connectors can override these
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://schema-registry:8081

plugin.path=/usr/share/kafka/plugins

The three internal topics — config, offset, and status storage — are what make distributed mode genuinely distributed: every worker in the cluster reads from these same topics, so any worker can answer a REST request about any connector's configuration or status, and a newly-joined worker picks up the full picture of what the cluster is currently running just by consuming these topics from the beginning, with no manual state transfer needed.

When Connect Is the Right Tool — and When It Is Not

Connect is a force multiplier for standard integrations, but it is not a universal replacement for application code that talks to Kafka. The deciding question is whether the integration is "move data from system A to system B, shaped roughly as-is" or whether it requires real business logic that a generic connector and a short SMT chain cannot express.

Reach for Kafka Connect when

• ✓A mature, well-maintained connector already exists for the system you need to integrate with — writing a bespoke integration for a solved problem is wasted engineering time.
• ✓The transformation needed is mechanical: renaming fields, masking a column, adding a static field, routing by a simple rule — squarely within what an SMT chain can express.
• ✓You want the integration to be config-driven and REST-managed rather than a deployable application — useful for teams that want data engineers, not software engineers, to own and adjust connectors.
• ✓You need the fault tolerance and rebalancing that a Connect cluster gives you for free, without writing that coordination logic yourself.

Write a custom producer or consumer when

• ✓The integration requires real business logic — conditional processing, calling another service mid-pipeline, joining against data that is not simply another Kafka topic or the source database itself.
• ✓No connector exists for the system, and building one would take longer than a purpose-built application, particularly for a one-off or low-volume integration.
• ✓You need fine-grained control over batching, partitioning strategy, or error handling that goes beyond what a connector's configuration surface exposes.
• ✓The integration is really a stream processing job — aggregating, joining, or windowing events — which belongs in Kafka Streams (Module 14), not in a source or sink connector.

Need: stream Postgres 'orders' table changes into Kafka.
  -> Debezium source connector. Solved problem, mature connector,
     mechanical mapping (row change -> Kafka record). Connect wins.

Need: land order events into S3 as Parquet, partitioned by date.
  -> S3 sink connector. Solved problem, standard partitioning
     behavior is exactly what the connector already does. Connect wins.

Need: when an order is marked "high fraud risk" by ML scoring AND
the customer has 2+ chargebacks in 90 days, call the risk API,
hold the order, and notify a Slack channel.
  -> Custom consumer. This is multi-step conditional business logic
     with an external API call and a lookup against historical state
     -- far beyond an SMT chain, and not a stream processing
     aggregation either. Hand-written service wins.

Five Misconceptions About Kafka Connect

What This Looks Like on Day One

5 Interview Questions — With Complete Answers

Mistakes Teams Make Running Kafka Connect

Errors You Will Hit — And Exactly Why They Happen