--- sidebar_position: 1 --- # What is Kafka Connect? Kafka Connect is a framework part of the Kafka ecosystem. It's primarily used for building and managing data pipelines between Kafka and other data systems, like databases, key-value stores, search indices, and file systems. It acts as an integrating bridge facilitating the ingestion of data from external data sources into topics, or extract data from topics into external systems. Common Kafka Connect use case is orchestrating real-time streams of events from a data source to a target for analytics. By having Kafka sit between the systems, the total system becomes loosely coupled, meaning that you can easily switch out the source or target, or stream to multiple targets, for example. And if the system gets overwhelmed, Kafka can act as a buffer, absorbing the back-pressure. Another use case is using change data capture (CDC) to allow your relational technologies to send data through Kafka to technologies such as NoSQL stores, other event-driven platforms, or micro-services—letting you unlock static data. In these circumstances, Kafka can serve as a message broker as well as an independent system of record. ## Connector plugins In Kafka Connect, a **connector** is the software component that implements the Connect framework to define the integration logic for specific data systems. These connector plugins are reusable components that define how source connectors ought to capture data from data sources to a Kafka topic and also how sink connectors should copy data from Kafka topics to be recognised by a target system. A connector can be a Sink -- if it ingests data into a system --, or a Source -- if it extracts data out of a system. A Connector will usually be easily configurable for a seamless integration with the Kafka ecosystem. In most cases it should remove the need for writing custom code for data integration with external systems. A **connector instance** is a logical job that is responsible of managing the copying of data between Kafka and another system. ## Tasks Each connector instance coordinates a set of tasks that copy data, which allows for parallelism and scalable data copying with minimal configuration. Tasks don't store any state withing them, a task's state is stored in special topics in Kafka (`config.storage.topic`, `status.storage.topic`) and managed by the connector. Tasks may be started, stopped, or restarted at any time. ## Workers But where do tasks actually run? Kafka Connect runs under the Java virtual machine (JVM) as a process known as a worker. In distributed mode each worker runs in its own Docker container. Each worker can run multiple connectors. When you add workers to a Kafka Connect cluster, the tasks are rebalanced across the available workers to distribute the workload. If you decide to scale down your cluster (or even if something outside your control happens and a worker crashes), Kafka Connect will rebalance again to ensure that all the connector tasks are still executed. ![2 connect clusters of 2 workers each, running 2 connectors](./img/multiple-workers-clusters.png) ## Converters To be able to read or write data in Kafka, tasks must use a Converter to change the format of data from bytes to an internal data format and viceversa. By default the following converters are available: | Converter | Class | Description | |------------------------ |---------------------------------------------------------- |--------------------------------------------------------- | | **AvroConverter** | `io.confluent.connect.avro.AvroConverter` | use with Schema Registry | | **ProtobufConverter** | `io.confluent.connect.protobuf.ProtobufConverter` | use with Schema Registry | | **JsonSchemaConverter** | `io.confluent.connect.json.JsonSchemaConverter` | use with Schema Registry | | **JsonConverter** | `org.apache.kafka.connect.json.JsonConverter` | (without Schema Reggistry): use with structured data | | **StringConverter** | `org.apache.kafka.connect.storage.StringConverter` | simple string format | | **ByteArrayConverter** | `org.apache.kafka.connect.cconverters.ByteArrayConverter` | provides a "pass-through" option that does no conversion | For more information on setting `key` and `value` converters in a connector see [configuring key and value converters](https://docs.confluent.io/platform/current/connect/userguide.html#configuring-key-and-value-converters). ## Transforms Connectors can be set up with transformations to make simple modifications to individual messages. This is often convenient for minor data adjustments and event routing. A transform is a simple function that accepts one record as an input and outputs a modified record. All transforms provided by Kafka Connect perform simple but commonly useful modifications, and you can also implement your own Transfromation with custom logic and use it with any conenctor. For more information on out-of-the-box available transformations, check the [Single Message Transformations (SMTs) in Confluent Platform](https://docs.confluent.io/platform/current/connect/transforms/overview.html#connect-transforms-supported). ## Dead Letter Queue Dead Letter Queues (DLQs) are only applicable for Sink connectors. If an invalid record were to be processed by a Sink connector, the error is handled based on the connector's `errors.tolerance` configuration property, which can be set to `all` or `none` (default). When the `errors.tolerance` property is set to `none`, an invalid record causes the connector task to immediately fail and the connector goes into a faield state. To resolve this issue, you must review the Kafka Connect Worker log and do the following: 1. Examine what caused the failure. 2. Fix the issue. 3. Restart the connector. If `errors.tolerance` is set to `all`, all errors or invalid records are ignored and processing continues, without errors being written to the Connect Worker log, and to determine if records are failing, you must rely on internal metrics, or count the number of records at the soruce and compare with the number of records processed. You can also handle errors by routing all the invalid messages to a special topic that can be monitored when they happen: the DLQ. Once you have a DLQ topic created, you can set the following configuration properties in your connector instance to start forwarding error messages to it: ``` errors.tolerance = all errors.deadletterqueue.topic.name = errors.deadletterqueue.context.headers.enable = true ``` Now you can see failed messages in the topic, and explore it's headers to determine why the record failed.