Kafka Connect

The Kafka Connect section lists the connectors of every Kafka Connect cluster that works with this Kafka cluster, with their status and task counts, and lets you look into a connector's tasks and configuration, pause, stop, resume and restart connectors and tasks, create, edit and copy connectors, reset their offsets and delete them. KafkIO does all of this through the Kafka Connect REST API.

To open it, connect the cluster and click Kafka Connect under its name in the Kafka Clusters pane. The entry is greyed out until at least one Connect cluster is set up in the cluster's settings (in the screenshot below, Storefront Prod (eu-west-1) has none).

1 2 3 4 5 6 7 8 9 10 11 12

1. Create Connector: start a new connector (below).
2. Connect Clusters: how many Connect clusters are set up for this Kafka cluster.
3. Failed Connectors: connectors whose own status is FAILED.
4. Failed Tasks: failed tasks across all connectors. This one can be above zero while every connector says RUNNING; see Status and tasks.
5. Filter connectors... (below).
6. The buttons that act on the selected connector (below).
7. Refresh All Connectors.
8. Connect Cluster: which Connect cluster runs the connector.
9. Refresh this connector only.
10. Tasks: the selected connector's tasks (below).
11. Config: its configuration (below).
12. Duplicate Tab: open the section in a tab of its own, as was done here for Consumers (Storefront Dev (Consumers)), where sink connectors show up as consumer groups (Duplicating a tab).

One or several Connect clusters

You don't pick a Connect cluster before you start: KafkIO reads the connectors of every Connect cluster set up for the Kafka cluster and lists them together, with the Connect Cluster column 8 saying which one runs each connector. The names in that column are the ones you gave each Connect cluster in the cluster's settings (Naming a Connect cluster). With only one Connect cluster, the column simply shows its name on every row.

To look at one Connect cluster's connectors, type its name into the filter. Everything you do to a connector goes to the Connect cluster on its row, so two connectors with the same name on different Connect clusters don't get mixed up. You choose the Connect cluster only when you create a connector.

If a Connect cluster can't be reached, the others' connectors are still listed and an error appears for the one that failed (in the current version as an Uncaught exception dialog; the cause is in the logs). See Testing the connection for the usual causes.

The connectors list

Each row is one connector:

Status and tasks

A connector does its work through tasks, which Connect spreads over its workers: a sink connector with tasks.max set to 3 on a three-partition topic gets three tasks, each reading some of the partitions. The Status column is only about the connector's coordinating part, so a connector can be RUNNING while all its tasks have failed and nothing is flowing, as with payments-export-sink above. Read the task columns, or the Failed Tasks card 4, to see whether a connector is really working.

Filtering and refreshing

Type in Filter connectors... 5 to show only the connectors whose name, Connect cluster, status, type, topics or plugin class contain the text: FAILED, SOURCE, analytics-connect, jdbc or a topic name all work. Switch on the .* toggle for a regular expression (Filtering). Click a column header to sort, for example by Failed Tasks to bring problems to the top.

The list is read when you open the section. It isn't refreshed by itself, so changes made elsewhere (a task failing, someone pausing a connector, a deployment adding one) show when you refresh: Refresh All Connectors 7 reads every Connect cluster again, and the refresh icon at the end of a row 9 (or Refresh on its right-click menu) re-reads just that connector and its tasks, which is quicker with many connectors. Changes you make in KafkIO show straight away. To keep watching one connector, use the Tasks tab's auto-refresh.

Right-click a connector for the same actions as the buttons, plus Duplicate Config and copying the row or all rows as text (Copying to the clipboard).

Connector details

Select a connector to see it in the tabs below the list (greyed out until you do). Its name appears after the section title; click it to copy it. As in other sections, you can drag the divider above the tabs to give them more room, or undock them into a window of their own (Split panes and detail tabs).

Tasks tab

The Tasks tab lists the selected connector's tasks, read the first time you select it and again after anything you do to the connector.

1 2 3

1. The expander: a red bug icon when the task has an error, which opens the full stack trace below the row; a tick when it has none.
2. Worker: the Connect worker running the task, as host:port. Its log is where to look for more detail.
3. The task's full error trace. Select and copy from it as from any text box.

At the right of the same tab:

4 5 6 7

1. Restart Connector Task: restart this task only, after a confirmation (also on the row's right-click menu).
2. Auto-refresh, every 5, 10, 20, 30 or 60 seconds (5 by default; off until you tick it).
3. Refresh: read the tasks now.
4. Undock the tabs into a window of their own.

Each row shows the Task ID (numbered from 0), its State (RUNNING, PAUSED, FAILED, RESTARTING, UNASSIGNED), the Worker and, for a failed task, the start of the error on one line in Error. The first line of the trace is usually the one that matters (here, a sink that can't create its output file); a Caused by: further down often names the real cause.

Auto-refresh re-reads the selected connector's tasks, and its row's counts, at the interval you choose, for as long as the box is ticked: handy while you wait for a restarted task to come back or to fail again. It only follows the selected connector; the rest of the list still needs a refresh.

Config tab

The Config tab shows the selected connector's configuration as JSON, exactly as Connect returns it, including its name. It's read-only.

Above it, on the right:

1 2 3

1. Search the configuration: matches are highlighted, the arrows step through them, and the count says how many there are.
2. Edit Config: change it (below).
3. View: open it in a window of its own.

Connect may hide secrets here: values that come from a config provider (such as ${file:/etc/secrets.properties:password}) are shown as the reference, not the secret.

Opening a config in its own window

View 3 opens the configuration in a separate window titled Connector: name, with the Connect cluster and connector name at the top and a search box. It doesn't block the rest of KafkIO, so you can open several and compare two connectors side by side. The window opens rather narrow and cuts off its header: widen it to read the names.

1

1. Always on top: on by default, so the window stays above the main window. Click it to let the window go behind.

Connector lifecycle

The buttons above the list act on the selected connector. Each asks for confirmation first, then waits a few seconds for Connect to report the new state and refreshes the connector's row and tasks. Buttons that don't apply to the connector's current state are greyed out.

1 2 3 4 5 6 7 8 9

1. Start/resume Connector.
2. Pause Connector.
3. Stop Connector.
4. Restart Connector.
5. Restart All Connector Tasks.
6. Restart Failed Connector Tasks.
7. Reset Connector Offsets (below).
8. Edit Config (below).
9. Delete Connector (below).

To restart a single task, use the restart icon at the end of its row on the Tasks tab. Restarting a task that failed because of something outside it (a database that was down, a missing file or permission) only helps once that is fixed; otherwise it fails again with the same error.

Pausing and stopping last until someone resumes the connector: Connect remembers the state, even across worker restarts. A paused or stopped sink connector's consumer group falls behind meanwhile, which you can watch in Consumers (the group is connect- followed by the connector name).

Creating a connector

Click Create Connector 1. In the dialog, name the connector, choose the Connect cluster to run it, and write its configuration as JSON.

1 2 3 4 5 6

1. Name: the connector's name on the Connect cluster. Required, and it can't be changed later.
2. Connect Cluster: which Connect cluster runs the connector.
3. The configuration, as a JSON object.
4. Examples: replace the editor's content with a ready-made configuration.
5. JSON validation: whether the text is well-formed JSON.
6. Create Connector: create it.

Choosing the Connect cluster

Connect Cluster 2 lists the Connect clusters set up for this Kafka cluster, by name, and starts on the first. Pick the one that has the connector's plugin installed and should own it: each Connect cluster runs only its own connectors, and names only need to be unique within one Connect cluster.

Writing the configuration

The editor takes the connector's configuration properties as one flat JSON object, the same as the config part of a Connect REST request: connector.class, tasks.max, topics (sinks) or the source's settings, converters, transforms and so on. Write every value as a string ("tasks.max": "1"), as Connect does. A name property is optional, but if you include it, it must be exactly the Name 1 above, or Connect refuses the connector. Which properties exist depends on the plugin: see its documentation.

The editor highlights JSON syntax and has its own search box.

Starting from the built-in examples

The links under the editor 4 fill it with a typical configuration for a common connector, replacing whatever was there:

They're starting points, not working configurations: replace the placeholders (file paths, topic names, hosts, database names and passwords, bucket and region) with your own, and check the property names against the plugin version you run (the Debezium example uses Debezium 1.x names such as database.server.name, which Debezium 2.x replaced with topic.prefix). Only the FileStream connectors come with Kafka, and even they aren't on the plugin path of every Connect installation; the others are plugins that must be installed on the Connect workers, or Connect refuses the connector because it can't find the class.

JSON validation

Shortly after you stop typing, KafkIO checks that the text is well-formed JSON, and JSON validation 5 says success or failed @ line n, with an info icon whose pop-up gives the line, column and parser message. It doesn't check the properties themselves: Connect does that when you create the connector, and refuses it with an error if, for example, a required property is missing or the plugin isn't installed. Create Connector stays available while the JSON is invalid, but the connector won't be created until it's fixed.

After creating

When Connect accepts the connector, KafkIO waits a few seconds for it to start, shows Connector created, and adds it to the list. The dialog stays open and becomes Update Connector, so you can correct the configuration straight away if the new connector's tasks fail. If a connector with that name already exists on the chosen Connect cluster, or Connect rejects the configuration, an error says why and nothing is created.

Updating a connector's configuration

Select a connector and click Edit Config (the pencil above the list, on the Config tab, or on the right-click menu). The Update Connector dialog opens with the current configuration; its Name and Connect Cluster are fixed. Edit the JSON and click Update Connector.

The JSON you submit replaces the whole configuration: a property you delete from the editor is removed from the connector (and falls back to its default), it isn't kept. Connect then restarts the connector and its tasks with the new settings, and they carry on from their committed offsets. To start from the beginning instead, reset the offsets.

Duplicating a connector

To make a connector like an existing one, for another topic or table, or to copy it to another Connect cluster, right-click it and choose Duplicate Config. A Create Connector dialog opens with its configuration, its Connect cluster selected, and the name changed to name_copy, both in Name and in the JSON's name.

1 2 3

1. Name, with _copy added.
2. Connect Cluster: the original's, but you can choose another.
3. The same name in the JSON.

Give the copy a proper name, and change it in both places: if Name and the JSON's name differ, Connect refuses the connector with Connector name configuration (...) doesn't match connector name in the URL. Then change whatever should differ from the original, such as its topics, output file or table, before you click Create Connector: a copied sink with the same output would write the same data twice. The copy is a new connector with its own consumer group and offsets, so it starts from the beginning of its topics (or of its source), not where the original is.

Resetting connector offsets

A connector's offsets record how far it has got: for a sink, the committed offsets of its consumer group (connect- plus the connector name) on each topic partition; for a source, the position in the external system (a line in a file, a row ID or timestamp in a table, a database log position) that Connect keeps in its offsets topic. Reset Connector Offsets 7 deletes them, so that when you start the connector again it begins from scratch: a sink from the start of its topics (with Connect's default settings), a source from the beginning of its data.

1. Stop the connector

   Select it and click Stop Connector. Offsets can only be reset while a connector is STOPPED (paused isn't enough), so the button is greyed out until then.

2. Reset the offsets

   Click Reset Connector Offsets and confirm with Reset Offsets. A notification says Connector offsets reset.

3. Start it again

   Click Start/resume Connector. Its tasks start without stored offsets and reprocess everything.

Resetting needs Kafka Connect 3.6 or later (Confluent Platform 7.6 or later); older Connect clusters answer with an error, as they do if the connector isn't stopped (Could not reset offsets of connector ..., is it STOPPED?). To move a sink to a particular position rather than back to the start, stop it and change its consumer group's offsets in Consumers.

Deleting a connector

Select the connector, click Delete Connector 9 (or use the right-click menu) and confirm with Delete. Connect stops its tasks and removes it, and it disappears from the list. There's no undo: to get it back, you'd create it again, so copy its configuration from the Config tab first if you might need it.

Deleting removes the connector only. Its topics and the data in them stay, and so does a sink's connect-name consumer group (with its committed offsets), which a new connector with the same name would pick up from. Delete the group in Consumers if you want a clean start.

Diagnosing connector and task errors

When data stops flowing:

1. Find the failures. Check the Failed Connectors and Failed Tasks cards, refresh, and sort or filter the list by FAILED. Remember that a RUNNING connector can have failed tasks.
2. Read the error. Select the connector and expand the failed task's row on the Tasks tab. The trace says what went wrong: a missing file or table, a refused login, a record that couldn't be converted (often a converter or schema mismatch), a topic that doesn't exist. The Worker column tells you which worker's log has the surrounding detail.
3. Fix the cause. Correct the configuration with Edit Config (this restarts the connector and its tasks), or fix the external system.
4. Restart. Use Restart Failed Connector Tasks, or the restart icon on one task's row; use Restart Connector when the connector itself is FAILED. Tick Auto-refresh to watch whether the tasks stay RUNNING.
5. Check it's catching up. For a sink, watch its connect-name group's lag fall in Consumers (a duplicated tab keeps it beside the connector).

Errors from KafkIO's own requests (creating, updating, pausing and so on) come in an error dialog quoting Connect's answer, which usually says exactly what it objected to:

A 400 Bad Request or 409 Conflict means Connect didn't accept the request (an invalid configuration, a name that's taken, or a connector being rebalanced; try again a moment later for the last); 401 and 403 are login and permission problems on the Connect REST API; a timeout or connection error means the Connect cluster can't be reached. The request timeouts are under Kafka & Timeouts, and connection problems are covered in Kafka Connect Connections.

Related

• Kafka Connect Connections (setting up Connect clusters)
• Consumers (sink connectors' consumer groups and lag)
• Duplicating a tab
• Kafka & Timeouts
• Features