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

# Logging with Confluent Manager for Apache Flink

Logging is an important aspect of troubleshooting your Confluent Manager for Apache Flink® (CMF) applications.
This topic describes how to configure logging for your Flink applications deployed with CMF.

## Overview

CMF exposes all Flink logging configurations with the application and environment APIs.
For more information, see [API reference](../../clients-api/rest.md#cmf-api-ref).

Flink uses a [Log4j 2](https://logging.apache.org/log4j/2.12.x/) configuration file, which you can use to define custom appenders for sending logs to a
remote system or to define a log format compatible with your log ingestion system.
When defining custom appenders, you should package a custom Docker image containing
the required additional dependencies.

By default, Flink clusters deployed with CMF will log to standard out, which you can access
from the pod logs or the Flink Web UI.

You can either choose to configure the logging behavior for a specific Application or per Environment. When you
configure logging for an Environment, the same behavior is applied to all of the Applications it contains.
A logging setting on an Environment takes precedence over the same setting at the Application level.

The content that you set for `log4j-console.properties` replaces the whole
default Log4j 2 configuration file of the Flink cluster, so it must define every
appender and logger that you need. The following examples define a console
appender, which writes logs to standard out.

## Option 1: Configure Logging for an Application

The following code shows an example of configuration for an Application.

```yaml
---
apiVersion: cmf.confluent.io/v1
kind: FlinkApplication
metadata:
  name: basic-example
spec:
  image: confluentinc/cp-flink:1.19.1-cp1
  flinkVersion: v1_19
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "4"
  serviceAccount: flink
  logConfiguration:
    log4j-console.properties: |+
      rootLogger.level = DEBUG
      rootLogger.appenderRef.console.ref = ConsoleAppender
      appender.console.name = ConsoleAppender
      appender.console.type = CONSOLE
      appender.console.layout.type = PatternLayout
      appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
  jobManager:
    resource:
      memory: 1048m
      cpu: 1
  taskManager:
    resource:
      memory: 1048m
      cpu: 1
  job:
    jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
    state: running
    parallelism: 2
    upgradeMode: stateless
```

## Option 2: Configure Logging for an Environment

The following code shows an example of configuring logging at the Environment level:

```yaml
---
name: default
kubernetesNamespace: staging-shared
flinkApplicationDefaults:
  spec:
    # Custom logging configuration for all Applications in this environment
    logConfiguration:
      log4j-console.properties: |+
        rootLogger.level = DEBUG
        rootLogger.appenderRef.console.ref = ConsoleAppender
        appender.console.name = ConsoleAppender
        appender.console.type = CONSOLE
        appender.console.layout.type = PatternLayout
        appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n
```

## Related content

- [Get Help with Confluent Platform for Apache Flink](../../get-help.md#cpf-get-help)
- [Collect Metrics for Confluent Manager for Apache Flink](metrics.md#cmf-metrics)