Kafka Connect Concepts

Table of contents

Reading Time: 5 minutes

Kafka Connect is a framework to stream data into and out of Apache Kafka.

A few major concepts.

Connectors – the high-level abstraction that coordinates data streaming by managing tasks
Tasks – the implementation of that how data is copied from Kafka
Workers – the running processes that execute connectors and tasks
Converters – the code used to translate data between Connect and the system sending or receiving data
Transforms – simple logic to alter each message produced by or sent to a connector
Dead Letter Queue – how Connect handles connector errors

Connectors

Connectors in Kafka Connect define where the data should be copied to and from. A connector instance is a logical job that is responsible for managing the copying of data between Kafka and another system. All of the classes that implement or are used by a connector, are defined in a connector plugin. Both connector instances and connector plugins may be referred to as “connectors”, but it should always be clear from the context in which is being referred to (for example, “install a connector” refers to the plugin, and “check the status of a connector” refers to a connector instance).

We encourage users to leverage existing connectors. However, it is possible to write a new connector plugin from scratch. At a high level, a developer who wishes to write a new connector plugin follows the workflow below. Further information is available in the developer guide.

Tasks

Tasks are the main actor in the data model for Connect. Each connector instance coordinates a set of tasks that actually copy the data. By allowing the connector to break a single job into many tasks, Kafka Connect provides built-in support for parallelism and scalable data copying with very little configuration. These tasks have no state stored within them. Task state is stored in Kafka in special topics config.storage.topic and status.storage.topic and managed by the associated connector.

Task Rebalancing

When a connector is first submitted to the cluster, the workers rebalance the full set of connectors in the cluster and their tasks so that each worker has approximately the same amount of work. This same rebalancing procedure is also used when connectors increase or decrease the number of tasks they require.

Workers

Connectors and tasks are logical units of work. Kafka Connect calls these processes workers and has two types of workers: standalone and distributed.

Standalone Workers

Standalone mode is the simplest mode, where a single process is responsible for executing all connectors and tasks.

Since it is a single process, it requires minimal configuration. Standalone mode is convenient for getting started, during development, and in certain situations where only one process makes sense, such as collecting logs from a host.

Distributed Workers

Distributed mode provides scalability and automatic fault tolerance for Kafka Connect. In distributed mode, you start many worker processes using the same group.id and they automatically coordinate to schedule execution of connectors and tasks across all available workers. If you add a worker, shut down a worker, or a worker fails unexpectedly, the rest of the workers detect this and automatically coordinate to redistribute connectors and tasks across the updated set of available workers. Note the similarity to consumer group rebalance. Under the covers, connect workers are using consumer groups to coordinate and rebalance.

Converters

Converters are necessary to have a Kafka Connect deployment support a particular data format when writing to or reading from Kafka. Tasks use converters to change the format of data from bytes to a Connect internal data format and vice versa.

By default, Confluent Platform provides the following converters:

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 Registry): use with structured data
StringConverter org.apache.kafka.connect.storage.StringConverter: simple string format
ByteArrayConverter org.apache.kafka.connect.converters.ByteArrayConverter: provides a “pass-through” option that does no conversion

Transforms

Connectors can be configured with transformations to make simple and lightweight modifications to individual messages. This can be convenient for minor data adjustments and event routing. And multiple transformations can be chained together in the connector configuration.

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. Note that you can implement the Transformation interface with your own custom logic, package them as a Kafka Connect plugin, and use them with any connectors.

When transforms are used with a source connector. Kafka Connect passes each source record produced by the connector through the first transformation, which makes its modifications and outputs a new source record. This updated source record is passed to the next transform in the chain, which generates a new modified source record. This continues for the remaining transforms. The final updated source record was converted to the binary form and written to Kafka.

Transforms can also be used with sink connectors. Kafka Connect reads messages from Kafka and converts the binary representation to a sink record. If there is a transform, Kafka Connect passes the record through the first transformation, which makes its modifications and outputs a new, updated sink record.

Dead Letter Queue

An invalid record may occur for a number of reasons. One example is when a record arrives at the sink connector serialized in JSON format, but the sink connector configuration is expecting Avro format. When an invalid record cannot be processed by a sink connector, the error is handled based on the connector configuration property errors.tolerance.

There are two valid values for this configuration property: none (default) or all.

When errors.tolerance is set to none an error or invalid record causes the connector task to immediately fail and the connector goes into a failed state. To resolve this issue, you would need to review the Kafka Connect Worker log to find out what caused the failure, correct it, and restart the connector.

When errors.tolerance is set to all, all errors or invalid records are ignored and processing continues. To determine if records are failing you must use internal metrics or count the number of records at the source and compare that with the number of records processed.