# Troubleshoot Self-Managed Kafka Connect for Confluent Platform

Use this page to troubleshoot self-managed Kafka Connect. For the procedures
outlined in this page, the assumption is that your Connect worker and source
connector are running, but Kafka Connect is not ingesting any data.

## Get the connector and task status

One way to troubleshoot self-managed Kafka Connect is to retrieve the
connector and task status. If Kafka Connect is not ingesting any data, you can
check whether one or more of the connector’s tasks has failed. You can use the
Connect REST API to gather more details:

1. Get the overall status of the connector instance:
   ```bash
   curl -s "http://localhost:8083/connectors/jdbc-sink/status" | jq '.connector.state'
   ```

   You should see output similar to:
   ```text
   "RUNNING"
   ```
2. Get the status of the connector’s individual tasks (checking the first task, index 0, here):
   ```bash
   curl -s "http://localhost:8083/connectors/jdbc-sink/status" | jq '.tasks[0].state'
   ```

   If one of the connector’s tasks has failed, you might see output like:
   ```text
   "FAILED"
   ```
3. If a task state is `"FAILED"`, use the following commands to request the stack trace for that task (again, using index 0 for the first task) and format it for readability. This helps identify what caused the failure.
   ```bash
   # Get the raw trace string from the JSON response
   TRACE=$(curl -s "http://localhost:8083/connectors/jdbc-sink/status" | jq -r '.tasks[0].trace')
   # Print the trace, interpreting escaped newlines (\n) and tabs (\t)
   echo -e "$TRACE"
   ```

   You might see output similar to this example, indicating a missing JDBC driver:
   ```text
   org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
       at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:618)
       at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:334)
       at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:238)
       at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:207)
       at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:189)
       at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:238)
       at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
       at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
       at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
       at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
       at java.base/java.lang.Thread.run(Thread.java:829)
   Caused by: org.apache.kafka.connect.errors.ConnectException: java.sql.SQLException: No suitable driver found for jdbc:mysql://localhost/demo
       at io.confluent.connect.jdbc.util.CachedConnectionProvider.getConnection(CachedConnectionProvider.java:59)
       at io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:64)
       at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:84)
       at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:595)
       ... 10 more
   Caused by: java.sql.SQLException: No suitable driver found for jdbc:mysql://localhost/demo
       at java.sql/java.sql.DriverManager.getConnection(DriverManager.java:702)
       at java.sql/java.sql.DriverManager.getConnection(DriverManager.java:189)
       at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.getConnection(GenericDatabaseDialect.java:247)
       at io.confluent.connect.jdbc.util.CachedConnectionProvider.newConnection(CachedConnectionProvider.java:80)
       at io.confluent.connect.jdbc.util.CachedConnectionProvider.getConnection(CachedConnectionProvider.java:52)
       ... 13 more
   ```

   As you read through the previous stack trace, you should notice the following key information:
   - A `ConnectException` indicating the task failed.
   - The root cause (often revealed in the “Caused by:” sections), such as `java.sql.SQLException: No suitable driver found`.

## Use Kafka Connect Log4j2 logging

In addition to the stack trace from the REST API, reading the connector logs directly is crucial. Kafka Connect uses **Apache Log4j2** for logging. The configuration, typically found in the `connect-log4j2.yaml` file (which must use Log4j2 syntax internally), controls what is logged, the message layout (pattern), and where log messages are sent (appenders like console or files):

```text
/etc/kafka/connect-log4j2.yaml (default location, uses Log4j2 syntax)
```

While the Properties format is common, Log4j2 also supports other formats like YAML, XML, or JSON. Using these may require including additional dependencies (like Jackson for YAML/JSON) and specifying the configuration file via system properties when starting Connect. For more details, See [Kafka Connect Logging](/platform/current/connect/logging.html)
The following are different ways to access the log output, depending on how you are running Connect:

- **Confluent CLI (Local)**

  If you are running Confluent Platform locally using the Confluent CLI, use the following command:
  ```bash
  confluent local services connect log
  ```
- **Docker**

  If you are using Docker, use `docker logs` followed by the container name (which might be `kafka-connect` or similar depending on your setup):
  ```bash
  # Example using docker-compose setup
  docker-compose logs kafka-connect
  # Example targeting a specific container name
  docker logs <your-connect-container-name>
  ```
- **Direct File Access (Manual/Systemd)**

  If you are running vanilla Kafka Connect (for example, using startup scripts or systemd units) and have configured file logging in Log4j2, you can read the log files directly using standard Linux commands like `cat` or `tail`. The location varies by installation and Log4j2 configuration (for example, the fileName property in the config).
  ```bash
  # Example assuming default log location often used
  tail -f /var/log/kafka/connect.log
  # Or using journalctl if managed by systemd
  sudo journalctl -f -u confluent-connect
  ```

In [Apache Kafka 2.3](https://www.confluent.io/blog/whats-new-in-apache-kafka-2-3/), connector contexts were added to logging via [KIP-449](https://cwiki.apache.org/confluence/display/KAFKA/KIP-449%3A+Add+connector+contexts+to+Connect+worker+logs). This allows Log4j2 patterns (using `%X{connector.context}`) to include identifiers like `[connector-name|task-id]` in log messages, making diagnostics much easier.

### How to identify the cause of the problem

In the following Connect worker log example, the initial ERROR line (line 1) indicates a severe problem but doesn’t immediately reveal the root cause. When you see such generic failure messages, you must examine the accompanying stack trace within the log for the underlying issue.

```text
 [2022-07-19 23:57:28,600] ERROR [jdbc-sink|task-0] WorkerSinkTask{id=jdbc-sink-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:207)
 org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
     at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:618)
     at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:334)
     at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:238)
     at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:207)
     at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:189)
     at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:238)
     at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
     at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
     at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
     at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
     at java.base/java.lang.Thread.run(Thread.java:829)
 Caused by: org.apache.kafka.connect.errors.ConnectException: java.sql.SQLException: No suitable driver found for jdbc:mysql://localhost/demo
     at io.confluent.connect.jdbc.util.CachedConnectionProvider.getConnection(CachedConnectionProvider.java:59)
     at io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:64)
     at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:84)
     at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:595)
     ... 10 more
 Caused by: java.sql.SQLException: No suitable driver found for jdbc:mysql://localhost/demo
     at java.sql/java.sql.DriverManager.getConnection(DriverManager.java:702)
     at java.sql/java.sql.DriverManager.getConnection(DriverManager.java:189)
     at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.getConnection(GenericDatabaseDialect.java:247)
     at io.confluent.connect.jdbc.util.CachedConnectionProvider.newConnection(CachedConnectionProvider.java:80)
     at io.confluent.connect.jdbc.util.CachedConnectionProvider.getConnection(CachedConnectionProvider.java:52)
     ... 13 more
```

If you look further down into the Caused by: sections of the stack trace (highlighted lines 13 and 18 in this example), you can often pinpoint the original source of the problem. Here, it’s clearly stated: java.sql.SQLException: No suitable driver found for jdbc:mysql://localhost/demo.

Once you have identified a potential cause like this, you may need to do some research. In this case, consulting the [JDBC connector documentation](https://docs.confluent.io/kafka-connectors/jdbc/current/index.html) would reveal that the specific [JDBC driver for your database (MySQL in this case)](https://docs.confluent.io/kafka-connectors/jdbc/current/jdbc-drivers.html) must be downloaded and placed in a location included in the Connect worker’s plugin.path.

If your research doesn’t resolve the issue, consider seeking help from the [Confluent Community Forum](https://forum.confluent.io/), providing the relevant logs and configuration details.

## Configure the log dynamically

Dynamic log configuration, introduced via the REST API in [KIP-495](https://cwiki.apache.org/confluence/display/KAFKA/KIP-495%3A+Dynamically+Adjust+Log+Levels+in+Connect), allows you to change the logging level for specific Java packages or classes without restarting the Connect worker. This is extremely useful for targeted debugging.

For example, perhaps you suspect an issue within the JDBC connector’s code and want to enable highly detailed TRACE logging just for that connector, without flooding the logs with TRACE messages from the entire Connect framework.

1. **List the current logger levels:** You can see which loggers have explicitly set levels.
   ```bash
   curl -s http://localhost:8083/admin/loggers/ | jq
   ```

   Example output showing some default or previously set levels:
   ```json
   {
     "org.apache.kafka.connect.runtime.rest": {
       "level": "WARN"
     },
     "org.reflections": {
       "level": "ERROR"
     },
     "root": {
       "level": "INFO"
     }
   }
   ```
2. **Modify a specific logger’s level:** Use a PUT request to change the level for a specific logger (for example, the root package for the JDBC connector).
   ```bash
   # Set JDBC connector logging to TRACE level
   curl -s -X PUT -H "Content-Type:application/json" \
       http://localhost:8083/admin/loggers/io.confluent.connect.jdbc \
       -d '{"level": "TRACE"}' | jq '.'
   ```

   Successful response might like:
   ```json
   {
     "io.confluent.connect.jdbc": {
       "level": "TRACE"
     }
   }
   ```

#### NOTE
Remember that changes made via the REST API are temporary and only affect the worker node that processed the request. They do not modify the Log4j2 configuration file and will be reset upon worker restart.

## Related content

- Blog Post: [Kafka Connect Deep Dive – Error Handling and Dead Letter Queues](https://www.confluent.io/blog/kafka-connect-deep-dive-error-handling-dead-letter-queues/)
- [Troubleshoot the Oracle CDC Source Connector](https://docs.confluent.io/kafka-connectors/oracle-cdc/current/troubleshooting.html)
- [Kafka Connect Logging](/platform/current/connect/logging.html)