Configure Flink job log export - Realtime Compute for Apache Flink

Notes

After configuring log export to OSS, SLS, or Kafka, you must restart the job.
If you configure log export to another storage service but do not disable the log archiving feature, logs will continue to be saved to the OSS or fully managed storage configured for your workspace. If you disable log archiving, you can no longer view job logs in the Flink console.

On the deployment details page, in the log configuration section, use the toggle in the log archiving row to enable or disable this feature.
You can use ${secret_values.xxxx} in the log configuration to reference variables stored in variable management. For more information, see project variable.

Configure single job log export

Console

Navigate to the log configuration page of a deployment.
1. Log in to the Realtime Compute for Apache Flink console.
2. In the Actions column of the target workspace, click Console.
3. In the left-side navigation pane, click O&M > Deployments. Then, click the name of the target deployment.
4. On the Configuration tab, in the Logging section, click Edit on the right.
5. For Logging Profile, select Custom Template.
Click Copy from System Template > default.
Click Add Appender and select a destination.

Configure the parameters for the selected appender.

To route logs of different levels to different destinations, configure filtering rules for the appender. For more information, see Configure log filtering based on log levels.

SLS

Parameter	Description
name	A custom name for the appender.
type	The appender type. This value is fixed to `SLS`. Do not change this value.
pattern	The output format for logs. The default value is `%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n`, which generates log entries such as `2023-10-01 14:23:45,678{GMT+8} INFO com.example.MyClass - This is a test log message`.
flushIntervalSeconds	The interval, in seconds, at which log data is flushed to the destination.
flushIntervalEventCount	The number of log entries to batch before writing to the storage destination. Note If both `flushIntervalSeconds` and `flushIntervalEventCount` are configured, the write operation is triggered when either threshold is met, whichever comes first.
authenticationMode	AccessKey SLS Token Note If you select SLS Token, you can export logs only to a Logstore in a Project that is in the same region as your Flink workspace. This feature is supported only in VVR 11.5 and later.
project	The name of the SLS Project.
logStore	The name of the Logstore.
endpoint	The private endpoint for the SLS service region. For more information, see Endpoints.
accessKeyId	The AccessKey ID and AccessKey Secret of the account used for the SLS service. For information about how to obtain an AccessKey pair, see Obtain an AccessKey pair. To avoid the security risks of using plaintext keys, we recommend storing AccessKey values in namespace variables. For more information, see Namespace variables. Note If your SLS and Realtime Compute for Apache Flink services are in different Alibaba Cloud accounts, you must grant the Flink account permission to write to SLS. For details, see Create a custom RAM policy. The following provides policy examples: Unrestricted SLS scope `{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": [ "log:Get", "log:PostLogStoreLogs" ], "Resource": "" } ] }` Specific SLS resource scope `{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": [ "log:PostLogStoreLogs", "log:GetLogStore" ], "Resource": "acs:log:cn-beijing:152940222687****:project/test-vvp-sls/logstore/test-ltest" } ] }`
accessKeySecret

OSS

The default values for the numerical parameters are: flushIntervalSeconds is 10, flushIntervalEventCount is 100, and rollingBytes is 1048576.

Parameter	Description
name	A custom name for the appender.
type	The appender type. This value is fixed to `OSS`. Do not change this value.
pattern	The output format for logs. The default value is `%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n`.
baseUri	The name of the OSS bucket.
endpoint	The endpoint for the region of your OSS service. For more information, see Regions and endpoints. Use the endpoint from the Endpoint (region) column that corresponds to the Access from ECS over the VPC (internal network) row.
accessKeyId	The AccessKey ID and AccessKey Secret of the account used for the OSS service. For information about how to obtain an AccessKey pair, see Obtain an AccessKey pair. To avoid the security risks of using plaintext keys, we recommend storing AccessKey values in namespace variables. For more information, see Namespace variables. Note These parameters are required only if the destination OSS bucket is in a different Alibaba Cloud account than your Flink service. If they are in the same account, remove these parameters from the configuration.
accessKeySecret
flushIntervalSeconds	The interval, in seconds, at which log data is flushed to the destination.
flushIntervalEventCount	The number of log entries to batch before writing to the storage destination. Note If both `flushIntervalSeconds` and `flushIntervalEventCount` are configured, the write operation is triggered when either threshold is met, whichever comes first.
rollingBytes	The maximum size of a single log file in OSS. When this size is reached, subsequent data is written to a new log file.

Kafka

Note

Kafka clusters with Kerberos authentication enabled are not supported.

Prerequisites

Because the KafkaAppender plug-in is loaded by the Flink plug-in class loader, you must explicitly declare its package path in the Flink configuration. This allows the application to load the plug-in for the current deployment.

On the Deployments page, click the name of your deployment. On the Configuration tab, in the Additional Flink Configuration section, add the following parameter to the Other Configuration field.
```
plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
```

Configuration

Parameter	Description
name	A custom name for the appender.
type	The appender type. This value is fixed to `KafkaVVP`. Do not change this value.
pattern	The output format for logs. The default value is `%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n`.
bootstrap.servers	The addresses of the Kafka brokers.
acks	The number of partition replicas that must receive a message before the producer considers the write successful. For more information, see acks.
buffer.memory	The total memory, in bytes, that the producer can use to buffer records.
retries	The number of times the producer will retry sending a message after a transient failure.
compression.type	The compression type for data generated by the producer. Valid values: `none`, `gzip`, `snappy`, `lz4`, and `zstd`.

Click Save.
At the top of the page, click Start.

XML method

To configure logging for a deployment, go to its log configuration page.
1. Log on to the Realtime Compute for Apache Flink console.
2. In the Actions column of the target workspace, click Console.
3. In the left-side navigation pane, click O&M > Deployments, and then click the name of the target deployment.
4. On the Deployment Details tab, in the Logging section, click Edit.
5. Set Logging Profile to Custom Template.

Configure the log output.

Depending on your storage destination, copy the corresponding configuration, paste it into the input box, and update the parameters with your storage details. To route logs of different levels to different destinations, you can configure filtering rules for each Appender. For more information, see Configure log outputs for different log levels.

OSS

<?xml version="1.0" encoding="UTF-8"?>
<Configuration xmlns="http://logging.apache.org/log4j/2.0/config" 
strict="true" packages="com.ververica.platform.logging.appender" status="WARN">  
  <Appenders> 
    <Appender name="StdOut" type="Console"> 
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout" charset="UTF-8"/> 
    </Appender> 
    <Appender name="RollingFile" type="RollingFile" fileName="${sys:log.file}" filePattern="${sys:log.file}.%i"> 
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout" charset="UTF-8"/>  
      <Policies> 
        <SizeBasedTriggeringPolicy size="20 MB"/> 
      </Policies>  
      <DefaultRolloverStrategy max="4"/> 
    </Appender>  
    <Appender name="OSS" type="OSS">
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout" charset="UTF-8"/>  
      <!-- The final effective log path is: ${baseUri}/logs/${namespace}/${deploymentId}/{jobId}/ -->
      <Property name="namespace">{{ namespace }}</Property> <!-- Do not modify this line -->
      <Property name="baseUri">oss://YOUR-BUCKET-NAME/</Property>
      <Property name="endpoint">https://YOUR-ENDPOINT</Property> 
      <Property name="accessKeyId">${secret_values.accessKeyId}</Property>
      <Property name="secretAccessKey">${secret_values.accessKeySecret}</Property>
      <Property name="flushIntervalSeconds">10</Property>  
      <Property name="flushIntervalEventCount">100</Property>  
      <Property name="rollingBytes">10485760</Property>  
    </Appender>
   <Appender name="StdOutErrConsoleAppender" type="Console">
     <Layout pattern="%m" type="PatternLayout" charset="UTF-8"/>
   </Appender>
   <Appender name="StdOutFileAppender" type="RollingFile" fileName="${sys:stdout.file}" filePattern="${sys:stdout.file}.%i">
     <Layout pattern="%m" type="PatternLayout" charset="UTF-8"/>
     <Policies>
     <SizeBasedTriggeringPolicy size="1 GB"/>
     </Policies>
     <DefaultRolloverStrategy max="2"/>
   </Appender>
   <Appender name="StdErrFileAppender" type="RollingFile" fileName="${sys:stderr.file}" filePattern="${sys:stderr.file}.%i">
     <Layout pattern="%m" type="PatternLayout" charset="UTF-8"/>
     <Policies>
     <SizeBasedTriggeringPolicy size="1 GB"/>
     </Policies>
     <DefaultRolloverStrategy max="2"/>
   </Appender>
  </Appenders>  
  <Loggers> 
    <Logger level="INFO" name="org.apache.hadoop"/>  
    <Logger level="INFO" name="org.apache.kafka"/>  
    <Logger level="INFO" name="org.apache.zookeeper"/>  
    <Logger level="INFO" name="akka"/>  
    <Logger level="ERROR" name="org.jboss.netty.channel.DefaultChannelPipeline"/>  
    <Logger level="OFF" name="org.apache.flink.runtime.rest.handler.job.JobDetailsHandler"/> 
    <Logger level="ERROR" name="org.apache.flink.fs.osshadoop.shaded.com.aliyun.oss"/>
  <Logger level="INFO" name="StdOutErrRedirector.StdOut" additivity="false">
    <AppenderRef ref="StdOutFileAppender"/>
    <AppenderRef ref="StdOutErrConsoleAppender"/>
  </Logger>
  <Logger level="INFO" name="StdOutErrRedirector.StdErr" additivity="false">
    <AppenderRef ref="StdErrFileAppender"/>
    <AppenderRef ref="StdOutErrConsoleAppender"/>
  </Logger>
    {%- for name, level in userConfiguredLoggers -%} 
      <Logger level="{{ level }}" name="{{ name }}"/> 
    {%- endfor -%}
    <Root level="{{ rootLoggerLogLevel }}"> 
      <AppenderRef ref="StdOut"/>
      <AppenderRef ref="RollingFile"/>  
      <AppenderRef ref="OSS"/> 
    </Root>
  </Loggers> 
</Configuration>

Parameter	Description
YOUR-BUCKET-NAME	Replace this with your OSS bucket name.
YOUR-ENDPOINT	Replace this with your OSS endpoint. For more information, see Regions and endpoints. This is the value from the Endpoint column in the Access from ECS over the VPC (internal network) row.
YOUR-OSS-ACCESSKEYID	Use the AccessKey ID and AccessKey secret for your OSS account. For more information about how to obtain them, see Obtain an AccessKey pair. To avoid exposing credentials in plaintext, this example uses namespace variables to specify the AccessKey values. For more information, see Namespace variables. Note These parameters are required only if the OSS bucket and your Realtime Compute for Apache Flink service are in different accounts. If the bucket is in the same account, you can omit these parameters.
YOUR-OSS-ACCESSKEYSECRET
flushIntervalSeconds	The interval at which log data is written to the storage destination, in seconds.
flushIntervalEventCount	The number of log entries to buffer before data is written to the destination. Note If both `flushIntervalSeconds` and `flushIntervalEventCount` are configured, the write operation is triggered when either threshold is met.
rollingBytes	The maximum size for a single log file in OSS. When a file reaches this size, a new log file is created for subsequent logs.

SLS

<?xml version="1.0" encoding="UTF-8"?>
<Configuration xmlns="http://logging.apache.org/log4j/2.0/config" 
strict="true" packages="com.ververica.platform.logging.appender" status="WARN">  
  <Appenders> 
    <Appender name="StdOut" type="Console"> 
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout" charset="UTF-8"/> 
    </Appender>  
    <Appender name="RollingFile" type="RollingFile" fileName="${sys:log.file}" filePattern="${sys:log.file}.%i"> 
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout" charset="UTF-8"/>  
      <Policies> 
        <SizeBasedTriggeringPolicy size="5 MB"/> 
      </Policies>  
      <DefaultRolloverStrategy max="1"/> 
    </Appender>  
    <Appender name="SLS" type="SLS">
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout" charset="UTF-8"/>  
      <!-- The final effective log path is: ${baseUri}/logs/${namespace}/${deploymentId}/{jobId}/ -->
      <Property name="namespace">{{ namespace }}</Property> <!-- Do not modify this line -->
      <Property name="project">YOUR-SLS-PROJECT</Property>  
      <Property name="logStore">YOUR-SLS-LOGSTORE</Property> 
      <Property name="endpoint">YOUR-SLS-ENDPOINT</Property> 
      <Property name="accessKeyId">${secret_values.accessKeyId}</Property> 
      <Property name="accessKeySecret">${secret_values.accessKeySecret}</Property> 
      <Property name="topic">{{ namespace }}:{{ deploymentId }}:{{ jobId }}</Property>
      <Property name="deploymentName">{{ deploymentName }}</Property>
      <Property name="flushIntervalSeconds">10</Property>
      <Property name="flushIntervalEventCount">100</Property>
    </Appender>
   <Appender name="StdOutErrConsoleAppender" type="Console">
     <Layout pattern="%m" type="PatternLayout" charset="UTF-8"/>
   </Appender>
   <Appender name="StdOutFileAppender" type="RollingFile" fileName="${sys:stdout.file}" filePattern="${sys:stdout.file}.%i">
     <Layout pattern="%m" type="PatternLayout" charset="UTF-8"/>
     <Policies>
     <SizeBasedTriggeringPolicy size="1 GB"/>
     </Policies>
     <DefaultRolloverStrategy max="2"/>
   </Appender>
   <Appender name="StdErrFileAppender" type="RollingFile" fileName="${sys:stderr.file}" filePattern="${sys:stderr.file}.%i">
     <Layout pattern="%m" type="PatternLayout" charset="UTF-8"/>
     <Policies>
     <SizeBasedTriggeringPolicy size="1 GB"/>
     </Policies>
     <DefaultRolloverStrategy max="2"/>
   </Appender>
  </Appenders>  
  <Loggers> 
    <Logger level="INFO" name="org.apache.hadoop"/>  
    <Logger level="INFO" name="org.apache.kafka"/>  
    <Logger level="INFO" name="org.apache.zookeeper"/>  
    <Logger level="INFO" name="akka"/>  
    <Logger level="ERROR" name="org.jboss.netty.channel.DefaultChannelPipeline"/>  
    <Logger level="OFF" name="org.apache.flink.runtime.rest.handler.job.JobDetailsHandler"/> 
    <Logger level="ERROR" name="org.apache.flink.fs.osshadoop.shaded.com.aliyun.oss"/>
  <Logger level="INFO" name="StdOutErrRedirector.StdOut" additivity="false">
    <AppenderRef ref="StdOutFileAppender"/>
    <AppenderRef ref="StdOutErrConsoleAppender"/>
  </Logger>
  <Logger level="INFO" name="StdOutErrRedirector.StdErr" additivity="false">
    <AppenderRef ref="StdErrFileAppender"/>
    <AppenderRef ref="StdOutErrConsoleAppender"/>
  </Logger>
    {%- for name, level in userConfiguredLoggers -%} 
      <Logger level="{{ level }}" name="{{ name }}"/> 
    {%- endfor -%}
    <Root level="{{ rootLoggerLogLevel }}"> 
      <AppenderRef ref="StdOut"/>
      <AppenderRef ref="RollingFile"/>  
      <AppenderRef ref="SLS"/> 
    </Root>
  </Loggers> 
</Configuration>

Note

The namespace, deploymentId, jobId, and deploymentName properties in the code are Twig variables. Do not modify them. Otherwise, the job fails to start.

Parameter	Description
YOUR-SLS-PROJECT	Replace this with your SLS Project name.
YOUR-SLS-LOGSTORE	Replace this with your SLS Logstore name.
YOUR-SLS-ENDPOINT	Replace this with the private endpoint for your SLS service's region. For more information, see Service endpoints.
YOUR-SLS-ACCESSKEYID	Use the AccessKey ID and AccessKey secret for your SLS account. For more information about how to obtain them, see Obtain an AccessKey pair. To avoid exposing credentials in plaintext, this example uses namespace variables to specify the AccessKey values. For more information, see Namespace variables. Note If your SLS and Realtime Compute for Apache Flink services are in different accounts, you must grant the Flink account write permissions to SLS. For more information, see Create a custom RAM policy. Use one of the following policies: Unrestricted SLS scope `{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": [ "log:Get", "log:PostLogStoreLogs" ], "Resource": "" } ] }` Scoped to a specific SLS resource. Example: `{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": [ "log:PostLogStoreLogs", "log:GetLogStore" ], "Resource": "acs:log:cn-beijing:152940222687****:project/test-vvp-sls/logstore/test-ltest" } ] }`
YOUR-SLS-ACCESSKEYSECRET
flushIntervalSeconds	The interval at which log data is written to the storage destination, in seconds.
flushIntervalEventCount	The number of log entries to buffer before data is written to the destination. Note If both `flushIntervalSeconds` and `flushIntervalEventCount` are configured, the write operation is triggered when either threshold is met.

Kafka

Note

Kerberos authentication for Kafka clusters is not currently supported.

Prerequisites

Realtime Compute for Apache Flink provides a KafkaAppender logging plugin. Because Flink loads this plugin by using a plug-in class loader, you must explicitly specify its package path so that your application can load it successfully. You can do this in one of the following ways:
- Configure a job template (affects all deployments in this workspace)
  
  On the Configurations page of the console, add the following code to the other configuration field.
```
plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
```
- Configure a single deployment (affects only the current deployment)
  
  On the Deployments page, click the name of the target deployment. On the Deployment Details tab, in the Parameters section, add the following code to the other configuration field.
```
plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
```

Logging configuration

<?xml version="1.0" encoding="UTF-8"?>
<Configuration xmlns="http://logging.apache.org/log4j/2.0/config" 
strict="true" packages="com.ververica.platform.logging.appender" status="WARN">  
  <Appenders> 
    <Appender name="StdOut" type="Console"> 
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout"/> 
    </Appender>  
    <Appender name="RollingFile" type="RollingFile" fileName="${sys:log.file}" filePattern="${sys:log.file}.%i"> 
      <Layout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" type="PatternLayout"/>  
      <Policies> 
        <SizeBasedTriggeringPolicy size="20 MB"/> 
      </Policies>  
      <DefaultRolloverStrategy max="4"/> 
    </Appender>  
    <Appender type="KafkaVVP" name="KafkaVVPAppender" topic="YOUR-TOPIC-NAME">
        <Layout type="PatternLayout" pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n"/>
        <Property name="bootstrap.servers">YOUR-KAFKA-BOOTSTRAP-SERVERS</Property>
         <Property name="acks">YOUR-ACKS-VALUE</Property>
         <Property name="buffer.memory">YOUR-BUFFER-MEMORY-SIZE</Property>
          <Property name="retries">YOUR-RETRIES-NUMBER</Property>
         <Property name="compression.type">YOUR-COMPRESSION-TYPE</Property>
    </Appender>
    <Appender type="Async" name="AsyncAppender">
        <AppenderRef ref="KafkaVVPAppender"/>
    </Appender>
  </Appenders>
  <Loggers> 
    <Logger level="INFO" name="org.apache.hadoop"/>  
    <Logger level="INFO" name="org.apache.kafka"/>  
    <Logger level="INFO" name="org.apache.zookeeper"/>  
    <Logger level="INFO" name="akka"/>  
    <Logger level="ERROR" name="org.jboss.netty.channel.DefaultChannelPipeline"/>  
    <Logger level="OFF" name="org.apache.flink.runtime.rest.handler.job.JobDetailsHandler"/> 
    {%- for name, level in userConfiguredLoggers -%} 
      <Logger level="{{ level }}" name="{{ name }}"/> 
    {%- endfor -%}
    <Root level="{{ rootLoggerLogLevel }}"> 
      <AppenderRef ref="StdOut"/>
      <AppenderRef ref="RollingFile"/>  
      <AppenderRef ref="AsyncAppender"/> 
    </Root>
  </Loggers>
</Configuration>

Parameter	Description
YOUR-TOPIC-NAME	The name of the destination Kafka topic.
YOUR-KAFKA-BOOTSTRAP-SERVERS	The addresses of the destination Kafka brokers.
YOUR-ACKS-VALUE	The number of partition replicas that must acknowledge a message before the producer considers the write successful. For more information, see acks.
YOUR-BUFFER-MEMORY-SIZE	The producer buffer size, in bytes.
YOUR-RETRIES-NUMBER	The number of retries after a send failure.
YOUR-COMPRESSION-TYPE	The compression type the producer uses for messages. Valid values include none, gzip, snappy, lz4, and zstd.

Click Save.
Click Start at the top of the page.

Configure log export for a namespace

You can use a template to configure the default log export destination to OSS, SLS, or Kafka for all jobs within a namespace.

Important

Once you configure this template, logs from all new jobs in this namespace are exported to OSS, SLS, or Kafka.
The Flink plugin class loader loads the KafkaAppender log plugin provided by Realtime Compute for Apache Flink. Before you can use this plugin, you must specify its class path to allow your Flink application to load it. To do this, in the Realtime Compute for Apache Flink development console, go to the Configurations page and add the following code to the Other Configuration section.
```
plugin.classloader.parent-first-patterns.additional: com.ververica.platform.logging.appender
```

Go to the job log template configuration page.
1. Log in to the Realtime Compute for Apache Flink console.
2. In the Actions column of the target workspace, click Console. In the Realtime Compute for Apache Flink development console, select the target namespace from the top of the page.
3. In the left-side navigation pane, click O&M > Configurations.
4. On the Deployment Defaults tab, select a job type.
5. In the Logging section, set Logging Profile to Custom Template.
Configure the log export destination for all jobs in the namespace.

For the required code, see Configure log export for a single job (XML).
Click save changes.

Configure level-based log outputs

You can use the Log4j2 ThresholdFilter to configure different filtering rules for each appender based on log level. This configuration offers several advantages:

Flexibility: You can set different log levels for different external storage destinations.
Efficiency: Reduces unnecessary log processing and transfer, improving system performance.
Clarity: Separate configurations clarify the log flow and simplify level management.

Follow these steps:

In the logging section, set Logging Profile to Custom Template.

Configure the log output.

The following example shows how to configure the system to send logs of INFO level and higher to the Realtime Compute for Apache Flink development console, while sending only logs of ERROR level and higher to Simple Log Service (SLS).

<?xml version="1.0" encoding="UTF-8"?>
<Configuration xmlns="http://logging.apache.org/log4j/2.0/config" strict="true" status="WARN">
  <Appenders>
    <!-- Console Appender configured to output logs of INFO level and higher -->
    <Appender name="StdOut" type="Console">
      <ThresholdFilter level="INFO" onMatch="ACCEPT" onMismatch="DENY"/>
      <Layout type="PatternLayout" pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" charset="UTF-8"/>
    </Appender>
    <!-- RollingFile Appender logs messages at the root logger's level (INFO) and higher -->
    <Appender name="RollingFile" type="RollingFile">
      <!-- Configuration remains unchanged -->
      <!-- ... -->
    </Appender>
    <!-- SLS Appender configured to output only ERROR level and above logs -->
    <Appender name="SLS" type="SLS">
      <ThresholdFilter level="ERROR" onMatch="ACCEPT" onMismatch="DENY"/>
      <Layout type="PatternLayout" pattern="%d{yyyy-MM-dd HH:mm:ss,SSS}{GMT+8} %-5p %-60c %x - %m%n" charset="UTF-8"/>
      <!-- SLS specific properties -->
      <Property name="namespace">YOUR_NAMESPACE</Property>
      <Property name="project">YOUR_SLS_PROJECT</Property>
      <Property name="logStore">YOUR_SLS_LOGSTORE</Property>
      <Property name="endpoint">YOUR_SLS_ENDPOINT</Property>
      <!-- Access credentials and other properties -->
      <!-- ... -->
    </Appender>
    <!-- Other Appenders definitions remain unchanged -->
    <!-- ... -->
  </Appenders>
  <Loggers>
    <!-- Define specific loggers that use the StdOut and SLS appenders -->
    <Logger name="StdOutLogger" level="INFO" additivity="false">
      <AppenderRef ref="StdOut"/>
    </Logger>
    <Logger name="SLSLogger" level="ERROR" additivity="false">
      <AppenderRef ref="SLS"/>
    </Logger>
    <!-- Other Loggers definitions with their specific configurations -->
    <!-- ... -->
    <!-- The Root logger sends output to StdOut and RollingFile, but not SLS, to avoid duplicate logs. -->
    <Root level="INFO">
      <AppenderRef ref="StdOut"/>
      <AppenderRef ref="RollingFile"/>
      <!-- Exclude SLS from Root to prevent duplicate logging in case of other loggers -->
    </Root>
  </Loggers>
</Configuration>

In this configuration:

Console Appender: Uses a ThresholdFilter to ensure that logs of the INFO level or higher are exported to the development console.
The SLS Appender uses a ThresholdFilter to send only logs of ERROR level and higher to SLS. For details on the SLS Appender properties, see Configure Output to SLS. Replace placeholders such as YOUR_NAMESPACE and YOUR_SLS_PROJECT with your actual SLS project information.

Note
If the SLS Appender type is not SLS but a custom appender, ensure that you use the correct type and that the appender class has the logic required to connect to SLS.
StdOutLogger and SLSLogger: They send logs only to the StdOut Appender and the SLS Appender, respectively, and each has its own log level restrictions.
Root Logger: This logger is configured with a StdOut Appender and a RollingFile Appender, but does not include an SLS Appender. This is to avoid sending duplicate logs to SLS if a specific logger is already configured to do so.

For more information about Log4j configuration parameters, see Apache Log4j.

Realtime Compute for Apache Flink:Configure job log export

Notes

Configure single job log export

Console

SLS

OSS

Kafka

XML method

OSS

SLS

Kafka

Configure log export for a namespace

Configure level-based log outputs

Related topics