<a id="connect-logging"></a>

# Kafka Connect Logging for Confluent Platform

Kafka Connect and other Confluent Platform components utilize the Java-based logging utility
[Apache Log4j2](https://logging.apache.org/log4j/2.x/) to collect runtime data
and record component events. Log4j2 can be configured using various formats, including Properties, XML, JSON, or YAML. While Confluent Platform often ships with a default `connect-log4j2.yaml` file, YAML offers a more structured and readable alternative.

#### IMPORTANT
Confluent Platform and its components, version 7.9 and earlier, use Log4j 1.x
and not Log4j 2.x. This affects how you configure logging and the
format of the logging configuration file. For information about
logging in Confluent Platform 7.9 and earlier, see
[Configure Confluent Platform Logging (7.8)](https://docs.confluent.io/platform/7.8/monitor/cp-logging.html).

The following table describes each standard log level:

| Level   | Description                                                                       |
|---------|-----------------------------------------------------------------------------------|
| OFF     | Turns off logging.                                                                |
| FATAL   | Severe errors that cause premature termination.                                   |
| ERROR   | Other runtime errors or unexpected conditions.                                    |
| WARN    | Runtime situations that are undesirable or unexpected, but not necessarily wrong. |
| INFO    | Runtime events of interest at startup and shutdown.                               |
| DEBUG   | Detailed diagnostic information about events.                                     |
| TRACE   | Detailed diagnostic information about everything, including fine-grained events.  |

By default, Connect typically writes `INFO`, `WARN`, `ERROR`, and `FATAL`
information to standard output (`stdout`) via a Console appender configured in its Log4j2 file. When Connect starts, it logs the settings its using and any `WARN` and `ERROR` (or `FATAL`) messages. To see detailed event flows, configure `DEBUG` or `TRACE` levels for specific components in the Log4j2 configuration file.

#### NOTE
To use a YAML configuration file (for example, `connect-log4j2.yaml`) with Kafka Connect:

1. Ensure the `jackson-dataformat-yaml` and its dependencies are included in Kafka Connect’s classpath.
2. Start Kafka Connect with the system property `log4j2.configurationFile` pointing to your YAML file. For example, by setting the KAFKA_OPTS environment variable before starting Connect:

```none
 export KAFKA_OPTS="-Dlog4j2.configurationFile=/path/to/etc/kafka/connect-log4j2.yaml"

(Adjust the path as necessary).
```

If this property is not set, Log4j2 will typically look for default configuration files like `log4j2.properties` or `log4j2.xml` on the classpath, or the explicitly provided `connect-log4j2.yaml` often used by Kafka Connect startup scripts.

## View Connect Logs

Kafka Connect typically writes logs to `stdout` by default, as configured in its Log4j2 properties. The following sections provide commands to view the Connect log output.

### Docker

Enter the following command to tail the Connect log for Confluent Platform running on
Docker:

```none
docker logs -f kafka-connect
```

For information about how to set up logging levels with Docker using environment variables, see
[Use environment variables (Docker)](#connect-logging-docker-env-vars).

### Confluent CLI

Enter the following command to display a snapshot of the log for Confluent Platform running locally:

```bash
confluent local services connect log
```

You can pipe the log through `grep` to filter for specific information. For
example, to see S3 connector messages, you could enter the following command:

```bash
confluent local services connect log | grep s3
```

When you run the standard log command, the location where logs are being directed (often `connect.stdout` or a file path if configured) might be indicated in startup messages or CLI output. You can use the `-f` flag to tail the log and display new messages as they are appended. For example:

```bash
confluent local services connect log -f
```

**Example output**: (Log4j2 format may vary based on pattern layout)

```none
[2023-10-27 10:30:15,123] DEBUG [s3-connector-task-0] Uploading part 1 for bucket 'my-s3-bucket', key 'topic/partition=0/0000000000.avro' (io.confluent.connect.s3.storage.S3OutputStream:286)
[2023-10-27 10:30:15,345] INFO [s3-connector-task-0] Successfully uploaded part 1, ETag: abcdef12345 (io.confluent.connect.s3.TopicPartitionWriter:410)
[2023-10-27 10:30:15,678] TRACE [s3-connector-task-0] Received record: SinkRecord{kafkaOffset=101, timestamp=1698382815678,...} (org.apache.kafka.connect.runtime.WorkerSinkTask:315)
```

<!-- 6.0 -- change/add ref link to 6.0 commands, for example :confluent-cli:`confluent local services connect log|command-reference/local/services/connect/confluent_local_services_connect_log.html` etc. -->

### Scripted Confluent Platform startup

When using [Confluent Platform systemd Service Unit Files](../installation/installing_cp/scripted-install.md#installing-systemd-unit)
to run Confluent Platform, Connect logs are typically managed by the `systemd` journal service. The default log file location might be `/var/log/confluent/connect` if file logging is explicitly configured in Log4j2 YAML, but often logs are primarily accessed via `journalctl`. Enter the
following `journalctl` commands to view log messages for Kafka Connect:

```none
sudo journalctl -u confluent-connect
```

Enter the following command to tail the log:

```none
sudo journalctl -f -u confluent-connect
```

## Log4j2 YAML Configuration File

A Log4j2 configuration file, typically named `connect-log4j2.yaml` if using YAML, defines how logging is handled. It consists of main sections like `Configuration`, `Appenders` (where logs go), and `Loggers` (which packages log at a certain level).

The following shows an example of a basic Log4j2 YAML configuration file for Kafka Connect:

```yaml
Configuration:
  status: warn # Log4j internal status logging level
  # dest: err # Optional: Where to send status logs (default is console)

  Appenders:
    # Console Appender (stdout)
    Console:
      name: stdout # Unique name for this appender
      PatternLayout:
        # The %X{connector.context} includes connector-specific context via MDC
        pattern: "[%d] %p %X{connector.context}%m (%c:%L)%n"

    # Rolling File Appender
    RollingFile:
      name: connectAppender # Unique name for this appender
      fileName: "${sys:kafka.logs.dir}/connect.log" # Base log file name. Using system property lookup.
      filePattern: "${sys:kafka.logs.dir}/connect-%d{yyyy-MM-dd-HH}.log.gz" # Pattern for rolled files, gzip compressed
      PatternLayout:
        pattern: "[%d] %p %X{connector.context}%m (%c:%L)%n"
      Policies:
        TimeBasedTriggeringPolicy:
          interval: 1 # Rollover interval based on the smallest unit in filePattern (HH=hourly)
          modulate: true # Align rollover to the interval boundary
      DefaultRolloverStrategy:
        max: 5 # Keep 5 rolled files

  Loggers:
    # Specific loggers to control noise
    Logger:
      - name: org.apache.zookeeper
        level: error
        additivity: false # Prevent logs from propagating to the root logger
      - name: org.reflections
        level: error
        additivity: false

    # Root logger: Default settings for all classes unless overridden above
    Root:
      level: info # Default log level
      AppenderRef:
        - ref: stdout # Send logs to the console appender by default
        # - ref: connectAppender # Uncomment to also send logs to the rolling file
```

You can customize this YAML file by adding more loggers, changing levels, or modifying appender settings.

## Enable file logging

To write logs to a file in addition to or instead of the console, uncomment the file appender reference in the Root logger’s `AppenderRef` section.

### Write logs to both console and file

The following example writes logs to both console and file:

```yaml
Root:
  level: info
  AppenderRef:
    - ref: stdout # Console output
    - ref: connectAppender # File output
```

### Write logs only to a file

To write logs only to a file and not to the console, remove the `stdout` reference:

```yaml
Root:
  level: info
  AppenderRef:
    - ref: connectAppender # File output only
```

The file appender configuration in the example provides the following capabilities:

* Writes logs to `${sys:kafka.logs.dir}/connect.log`.
* Rolls logs hourly (based on the `filePattern` with `HH`).
* Compresses rolled files (`.gz` extension).
* Keeps a maximum of 5 rolled files and automatically deletes older files.

## Customize file logging

You can customize the file appender configuration to match your operational requirements. The following patterns address common production scenarios.

### Daily log rotation with size-based triggering

The following example configures both time-based and size-based log rotation:

```yaml
RollingFile:
  name: connectAppender
  fileName: "${sys:kafka.logs.dir}/connect.log"
  filePattern: "${sys:kafka.logs.dir}/connect-%d{yyyy-MM-dd}-%i.log.gz"
  PatternLayout:
    pattern: "[%d] %p %X{connector.context}%m (%c:%L)%n"
  Policies:
    TimeBasedTriggeringPolicy:
      interval: 1
      modulate: true
    SizeBasedTriggeringPolicy:
      size: "100MB"
  DefaultRolloverStrategy:
    max: 30
```

This configuration provides the following capabilities:

* Rolls logs daily at midnight (`%d{yyyy-MM-dd}`).
* Rolls logs when the file reaches 100MB (`SizeBasedTriggeringPolicy`).
* Keeps up to 30 rolled files (approximately one month of daily logs).

### Separate log files for different log levels

The following example creates separate log files for different log levels:

```yaml
Appenders:
  RollingFile:
    name: errorFileAppender
    fileName: "${sys:kafka.logs.dir}/connect-error.log"
    filePattern: "${sys:kafka.logs.dir}/connect-error-%d{yyyy-MM-dd}.log.gz"
    PatternLayout:
      pattern: "[%d] %p %X{connector.context}%m (%c:%L)%n"
    ThresholdFilter:
      level: ERROR
      onMatch: ACCEPT
      onMismatch: DENY
    Policies:
      TimeBasedTriggeringPolicy:
        interval: 1
        modulate: true
    DefaultRolloverStrategy:
      max: 30

  RollingFile:
    name: infoFileAppender
    fileName: "${sys:kafka.logs.dir}/connect-info.log"
    filePattern: "${sys:kafka.logs.dir}/connect-info-%d{yyyy-MM-dd}.log.gz"
    PatternLayout:
      pattern: "[%d] %p %X{connector.context}%m (%c:%L)%n"
    Policies:
      TimeBasedTriggeringPolicy:
        interval: 1
        modulate: true
    DefaultRolloverStrategy:
      max: 7

Loggers:
  Logger:
    - name: org.apache.kafka.connect
      level: info
      additivity: false
      AppenderRef:
        - ref: infoFileAppender
        - ref: errorFileAppender
```

This configuration provides the following capabilities:

* Creates `connect-error.log` for ERROR and FATAL messages only (using `ThresholdFilter`).
* Creates `connect-info.log` for all log levels (INFO and above).
* Uses different retention policies: 30 days for errors, 7 days for info logs.
* Routes a specific logger (`org.apache.kafka.connect`) to both appenders.

### Set the log directory

The `${sys:kafka.logs.dir}` system property specifies where log files are written. You can set this property when starting Kafka Connect by using the `KAFKA_OPTS` environment variable:

```bash
export KAFKA_OPTS="-Dkafka.logs.dir=/var/log/kafka-connect -Dlog4j2.configurationFile=/path/to/connect-log4j2.yaml"
# Then start Connect
```

Alternatively, you can set this property in a properties file:

```none
log.dirs=/var/log/kafka-connect
```

## Change Log Levels

The following sections provide information on changing log levels using the Log4j2 YAML file or the REST API.

### Use the Connect Log4j2 YAML configuration file

To debug specific issues, you can adjust log levels within the `connect-log4j2.yaml` file. Setting targeted `DEBUG` or `TRACE` levels is usually more effective than setting the `Root` logger to DEBUG.

The following example YAML configuration sets `DEBUG` level for the Connect framework, worker tasks, and specific connectors (Datagen and S3), while keeping common dependencies quieter:

```yaml
Configuration:
  status: warn

  Appenders:
    Console:
      name: stdout
      PatternLayout:
        # Include connector context using MDC lookup %X{connector.context}
        pattern: "[%d] %p %X{connector.context}%m (%c:%L)%n"
    # Include RollingFile appender definition here if needed

  Loggers:
    # Suppress noisy logs from common dependencies
    Logger:
      - name: org.reflections
        level: error
        additivity: false
      - name: org.eclipse.jetty
        level: error
        additivity: false
      - name: org.apache.kafka # Covers Kafka clients and core classes
        level: error
        additivity: false
      - name: org.apache.zookeeper
        level: error
        additivity: false

      # Uncomment to debug Kafka clients (very verbose)
      # - name: org.apache.kafka.clients.consumer
      #   level: debug
      #   additivity: false
      # - name: org.apache.kafka.clients.producer
      #   level: debug
      #   additivity: false

      # Set DEBUG for source task execution:
      - name: org.apache.kafka.connect.runtime.WorkerSourceTask
        level: debug
        additivity: false # Usually false if parent 'o.a.k.connect' is also DEBUG

      # Set DEBUG for sink task execution:
      - name: org.apache.kafka.connect.runtime.WorkerSinkTask
        level: debug
        additivity: false

      # Set DEBUG for the general Connect framework (runtime, converters, SMTs):
      - name: org.apache.kafka.connect
        level: debug
        additivity: false # Set false to avoid duplicate messages from tasks if they inherit level

      # Enable DEBUG for specific connectors:
      - name: io.confluent.kafka.connect.datagen # Datagen connector
        level: debug
        additivity: false

      - name: io.confluent.connect.storage # Base storage package (S3, HDFS, GCS)
        level: debug
        additivity: false
      - name: io.confluent.connect.s3 # S3 specific classes
        level: debug
        additivity: false # Overrides storage if needed, prevents duplicates

      # Uncomment to enable DEBUG for other connectors:
      # - name: io.confluent.connect.jdbc
      #   level: debug
      #   additivity: false
      # - name: io.confluent.connect.elasticsearch
      #   level: debug
      #   additivity: false
      # - name: io.confluent.connect.hdfs
      #   level: debug
      #   additivity: false
      # - name: io.confluent.connect.gcs
      #   level: debug
      #   additivity: false
      # - name: io.confluent.connect.jms
      #   level: debug
      #   additivity: false

      # Add other connector-specific loggers here as needed following the pattern:
      # - name: <root.package.of.connector>
      #   level: debug # or trace
      #   additivity: false

    # Root logger catches everything else
    Root:
      level: info # Default level for unspecified loggers
      AppenderRef:
        - ref: stdout # Send root logs to the console
```

<a id="connect-logging-using-api"></a>

### Use the Connect API

After Confluent Platform, Kafka Connect, and your connectors are running, you can dynamically check and change log levels using the Connect Admin REST API endpoints for logging. This interacts with the active Log4j2 configuration in memory.

#### NOTE
Changes made through the API are temporary and affect only the specific worker node receiving the request. They do not modify the logging configuration file (`connect-log4j2.yaml`). When the worker restarts, logging configuration reverts to what is defined in the file. These API calls work regardless of the underlying configuration file format, as they target the logging facade (SLF4j).

The examples below assume a starting configuration similar to the second Log4j2 YAML example above (with specific DEBUG levels enabled).

#### Check all currently set log levels

Enter the following command to list all loggers with explicitly configured levels (including the root logger). Use `jq` for formatted JSON output.

```none
curl -Ss http://localhost:8083/admin/loggers | jq
```

**Example output**: Reflecting the example YAML configuration

```json
{
  "io.confluent.connect.s3": {
    "level": "DEBUG",
    "last_modified": 1700000000000
  },
  "io.confluent.connect.storage": {
    "level": "DEBUG",
    "last_modified": 1700000000000
  },
  "io.confluent.kafka.connect.datagen": {
    "level": "DEBUG",
    "last_modified": null
  },
  "org.apache.kafka": {
    "level": "ERROR",
    "last_modified": null
  },
  "org.apache.kafka.connect": {
    "level": "DEBUG",
    "last_modified": null
  },
  "org.apache.kafka.connect.runtime.WorkerSinkTask": {
    "level": "DEBUG",
    "last_modified": 1700150000000
  },
  "org.apache.kafka.connect.runtime.WorkerSourceTask": {
    "level": "DEBUG",
    "last_modified": null
  },
  "org.eclipse.jetty": {
    "level": "ERROR",
    "last_modified": null
  },
  "org.reflections": {
    "level": "ERROR",
    "last_modified": null
  },
  "root": {
    "level": "INFO",
    "last_modified": null
  }
}
```

#### Get the log level for a specific logger

Enter the following command to check the current log level specifically for the `WorkerSourceTask` logger:

```none
curl -Ss http://localhost:8083/admin/loggers/org.apache.kafka.connect.runtime.WorkerSourceTask | jq
```

**Example output:**

```json
{
  "level": "DEBUG",
  "last_modified": 1700000000000
}
```

#### Change the log level for a specific logger

Enter the following command to dynamically change the log level for the `WorkerSourceTask` logger from **DEBUG** to **TRACE**:

```none
curl -s -X PUT -H "Content-Type:application/json" \
http://localhost:8083/admin/loggers/org.apache.kafka.connect.runtime.WorkerSourceTask \
-d '{"level": "TRACE"}' | jq '.'
```

**Example output**: Shows the logger name that was modified

```json
{
  "org.apache.kafka.connect.runtime.WorkerSourceTask": {
    "level": "TRACE",
    "last_modified": 1700250000000
  }
}
```

#### Revert the log level for a specific logger

Enter the following command to change the log level back to **DEBUG** for the `WorkerSourceTask` logger:

```none
curl -s -X PUT -H "Content-Type:application/json" \
http://localhost:8083/admin/loggers/org.apache.kafka.connect.runtime.WorkerSourceTask \
-d '{"level": "DEBUG"}' | jq '.'
```

**Example output:**

```json
{
  "org.apache.kafka.connect.runtime.WorkerSourceTask": {
    "level": "DEBUG",
    "last_modified": 1700260000000
  }
}
```

#### NOTE
Remember, log levels set via the `/admin/loggers` REST API do not persist across worker restarts.

#### Change log levels cluster-wide

Enter the following command to change the log level for all workers in a distributed mode cluster. Use the `scope=cluster`
query parameter to broadcast the change to all workers.

```none
curl -s -X PUT -H "Content-Type:application/json" \
http://localhost:8083/admin/loggers/org.apache.kafka.connect.runtime.WorkerSinkTask?scope=cluster \
-d '{"level": "TRACE"}'
```

**Example output:**

```html
HTTP/1.1 204 No Content
```

When you use `scope=cluster`, the API returns `204 No Content` with an empty response body instead of the standard JSON response.
The log level adjustment is written to the configuration topic (`config.storage.topic`) and asynchronously propagated to all workers in the cluster. This includes:

* Workers currently running in the cluster
* Workers that start after the change is made
* Workers that rejoin the cluster after a temporary disconnection

#### NOTE
Cluster-wide log level changes may take some time to propagate to all workers, as each worker reads from the configuration topic independently. You can verify propagation by checking the `last_modified` timestamp using `GET /admin/loggers` on individual workers. The `scope=cluster` parameter is only meaningful in distributed mode; in standalone mode, it returns `204 No Content` but only affects the single worker instance.

To verify that the cluster-wide change has been applied, check the log levels on any worker:

```none
curl -Ss http://localhost:8083/admin/loggers/org.apache.kafka.connect.runtime.WorkerSinkTask | jq
```

**Example output:**

```json
{
  "level": "TRACE",
  "last_modified": 1700280000000
}
```

The `last_modified` timestamp helps you verify when each worker applied the cluster-wide change. All workers should eventually show similar timestamps after the propagation completes.

#### Change the default listener port for the admin API

[KIP-495](https://cwiki.apache.org/confluence/display/KAFKA/KIP-495%3A+Dynamically+Adjust+Log+Levels+in+Connect)
introduced the `/admin/loggers` REST API endpoint. The `admin.listeners` property in the worker configuration file controls the network interfaces and ports where this admin endpoint is accessible. By default, it shares the main REST API port (usually `8083`).

#### Change the log level for a connector package

Enter the following command to change the log level to **TRACE** for the entire package associated with the Amazon S3 sink connector:

```none
curl -s -X PUT -H "Content-Type:application/json" \
http://localhost:8083/admin/loggers/io.confluent.connect.s3 \
-d '{"level": "TRACE"}' | jq '.'
```

**Example output**: Log4j2 might only return the specific logger set.

```json
{
  "io.confluent.connect.s3": {
    "level": "TRACE",
    "last_modified": 1700270000000
  }
}
```

#### Change the default listener port for the admin API

[KIP-495](https://cwiki.apache.org/confluence/display/KAFKA/KIP-495%3A+Dynamically+Adjust+Log+Levels+in+Connect)
introduced the `/admin/loggers` REST API endpoint. The `admin.listeners` property in the worker configuration file controls the network interfaces and ports where this admin endpoint is accessible. By default, it shares the main REST API port (usually `8083`).

You can configure `admin.listeners` to expose the admin endpoints on a separate port, potentially secured with TLS, or disable them entirely. These changes are made in the Connect worker configuration file (`connect-distributed.properties` or `connect-standalone.properties`).

* To make the admin endpoints listen on a separate HTTP port (for example, `9093`):
  ```none
  admin.listeners=http://localhost:9093
  ```
* To set up a separate HTTPS listener for admin endpoints (for example, on port `9094`):
  ```none
  admin.listeners=https://localhost:9094
  admin.listeners.https.ssl.truststore.location=/path/to/truststore.jks
  admin.listeners.https.ssl.truststore.password=<truststore-password>
  admin.listeners.https.ssl.keystore.location=/path/to/keystore.jks
  admin.listeners.https.ssl.keystore.password=<keystore-password>
  # Add other necessary TLS/SSL properties (e.g., key password, protocol, cipher suites)
  ```
* To disable the admin endpoints entirely, set the property to an empty value:
  ```none
  admin.listeners=
  ```

<a id="connect-logging-docker-env-vars"></a>

### Use environment variables (Docker)

When running Kafka Connect in Docker containers provided by Confluent, you can override Log4j2 settings using environment variables. This is often easier than mounting custom configuration files (regardless of their format - Properties, YAML, etc.).

* To set the root logger level, use `CONNECT_LOG4J_ROOT_LOGLEVEL`:
  ```bash
  # Set root logger to DEBUG
  CONNECT_LOG4J_ROOT_LOGLEVEL=DEBUG
  ```
* To set levels for specific loggers, use `CONNECT_LOG4J_LOGGERS`. Provide a comma-separated list of `logger.name=LEVEL` pairs. Note the syntax matches Log4j2 logger naming conventions:
  ```bash
  # Enable DEBUG for the S3 connector package and TRACE for worker sink tasks
  CONNECT_LOG4J_LOGGERS="io.confluent.connect.s3=DEBUG,org.apache.kafka.connect.runtime.WorkerSinkTask=TRACE"
  ```
* To enable **DEBUG** log messages specifically for the Connect framework:
  ```bash
  CONNECT_LOG4J_LOGGERS="org.apache.kafka.connect=DEBUG"
  ```

Combine these variables as needed when starting your Confluent Platform Connect container. For example:

```bash
docker run -d \
  --name=kafka-connect \
  -p 8083:8083 \
  -e CONNECT_BOOTSTRAP_SERVERS=kafka:9092 \
  -e CONNECT_REST_ADVERTISED_HOST_NAME=kafka-connect \
  -e CONNECT_GROUP_ID=connect-cluster \
  # ... other required CONNECT_ variables ...
  -e CONNECT_LOG4J_ROOT_LOGLEVEL=INFO \
  -e CONNECT_LOG4J_LOGGERS="io.confluent.connect.s3=DEBUG,org.apache.kafka.connect.runtime=DEBUG" \
  confluentinc/cp-kafka-connect:latest
```

For additional Docker logging information specific to Confluent Platform images, see [Configure Kafka Logs for Docker in Confluent Platform](../installation/docker/operations/logging.md#docker-operations-logging).

## Stack Trace

If a connector task fails, you can retrieve the Java stack trace using the Connect status API endpoint. This provides detailed error information crucial for diagnosing the failure.

Enter the following command to get the status, including any task failures and traces, for a specific connector. Use `jq` to parse the JSON response.

```none
curl -s http://localhost:8083/connectors/<connector-name>/status | jq
```

For example, checking the status of a connector named `http-sink`:

```none
curl -s http://localhost:8083/connectors/http-sink/status | jq
```

**Example output**: Showing a failed task with a stack trace

```json
{
  "name": "http-sink",
  "connector": {
    "state": "RUNNING",
    "worker_id": "127.0.0.1:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "FAILED",
      "worker_id": "127.0.0.1:8083",
      "trace": "org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:618)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:334)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:238)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:207)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:189)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:238)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\tat java.base/java.lang.Thread.run(Thread.java:829)\nCaused by: org.apache.kafka.connect.errors.ConnectException: Connection error during HTTP POST request to target endpoint\n\tat io.confluent.connect.http.sink.HttpSinkWriter.write(HttpSinkWriter.java:95)\n\tat io.confluent.connect.http.sink.HttpSinkTask.put(HttpSinkTask.java:90)\n\tat org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:595)\n\t... 10 more\nCaused by: java.net.ConnectException: Connection refused\n\tat java.base/sun.nio.ch.Net.pollConnect(Native Method)\n\tat java.base/sun.nio.ch.Net.pollConnectNow(Net.java:669)\n\tat java.base/sun.nio.ch.NioSocketImpl.timedFinishConnect(NioSocketImpl.java:542)\n\tat java.base/sun.nio.ch.NioSocketImpl.connect(NioSocketImpl.java:597)\n\tat java.base/java.net.SocksSocketImpl.connect(SocksSocketImpl.java:333)\n\tat java.base/java.net.Socket.connect(Socket.java:645)\n\tat org.apache.http.conn.ssl.SSLConnectionSocketFactory.connectSocket(SSLConnectionSocketFactory.java:368)\n\tat org.apache.http.impl.conn.DefaultHttpClientConnectionOperator.connect(DefaultHttpClientConnectionOperator.java:142)\n\t... 18 more\n"
    }
  ],
  "type": "sink"
}
```

To extract and print just the stack trace in a more readable format (handling the escaped newline characters), you can combine `jq` with `echo -e` or process the string further:

```none
# Extract the trace for the first task (index 0)
TRACE=$(curl -s localhost:8083/connectors/<connector-name>/status | jq -r .tasks[0].trace)
# Print with interpreted newlines
echo -e "$TRACE"
```

For example:

```none
TRACE=$(curl -s localhost:8083/connectors/http-sink/status | jq -r .tasks[0].trace)
echo -e "$TRACE"
```

**Example output:**

```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: Connection error during HTTP POST request to target endpoint
    at io.confluent.connect.http.sink.HttpSinkWriter.write(HttpSinkWriter.java:95)
    at io.confluent.connect.http.sink.HttpSinkTask.put(HttpSinkTask.java:90)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:595)
    ... 10 more
Caused by: java.net.ConnectException: Connection refused
    at java.base/sun.nio.ch.Net.pollConnect(Native Method)
    at java.base/sun.nio.ch.Net.pollConnectNow(Net.java:669)
    at java.base/sun.nio.ch.NioSocketImpl.timedFinishConnect(NioSocketImpl.java:542)
    at java.base/sun.nio.ch.NioSocketImpl.connect(NioSocketImpl.java:597)
    at java.base/java.net.SocksSocketImpl.connect(SocksSocketImpl.java:333)
    at java.base/java.net.Socket.connect(Socket.java:645)
    at org.apache.http.conn.ssl.SSLConnectionSocketFactory.connectSocket(SSLConnectionSocketFactory.java:368)
    at org.apache.http.impl.conn.DefaultHttpClientConnectionOperator.connect(DefaultHttpClientConnectionOperator.java:142)
    ... 18 more
```

<!-- 6.0 -- add ref link to 6.0 commands, for example :ref:`confluent_local_services_connect_status` -->