Flink(1.12.1)日志配置Logback实现日志切分和kafka发送

<configuration>
    <property name="LOG_PATTERN"
              value="%d{yyyy-MM-dd HH:mm:ss.SSS} %app [%thread] %-5level %logger{60} %X{sourceThread} - %msg%n"/>

    <appender name="file" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>${log.file}</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
            <fileNamePattern>${log.file}.%i</fileNamePattern>
            <minIndex>1</minIndex>
            <maxIndex>10</maxIndex>
        </rollingPolicy>
        <triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
            <maxFileSize>10KB</maxFileSize>
        </triggeringPolicy>
        <encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder">
            <layout class="com.changan.carbond.common.logconver.AppNameLayOut">
                <pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} %app [%thread] %-5level %logger{60} %X{sourceThread} - %msg%n
                </pattern>
            </layout>
            <charset>UTF-8</charset>
        </encoder>
    </appender>

    <appender name="kafka" class="com.github.danielwegener.logback.kafka.KafkaAppender">

        <encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder">
            <layout class="com.changan.carbond.common.logconver.AppNameLayOut">
                <pattern>${LOG_PATTERN}</pattern>
            </layout>
            <charset>UTF-8</charset>
        </encoder>
        <topic>flink-app-logs</topic>
        <keyingStrategy class="com.github.danielwegener.logback.kafka.keying.NoKeyKeyingStrategy"/>
        <deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy"/>
        <producerConfig>bootstrap.servers=broker1:9092,broker2:9092,broker3:9092</producerConfig>
        <producerConfig>retries=3</producerConfig>
        <producerConfig>acks=1</producerConfig>
        <producerConfig>batch-size=16384</producerConfig>
        <producerConfig>buffer-memory=33554432</producerConfig>
        <!-- wait up to 1000ms and collect log messages before sending them as a batch -->
        <producerConfig>linger.ms=1000</producerConfig>
        <!-- even if the producer buffer runs full, do not block the application but start to drop messages -->
        <producerConfig>max.block.ms=0</producerConfig>
    </appender>

    <!-- This affects logging for both user code and Flink -->
    <root level="INFO">
        <appender-ref ref="file"/>
        <appender-ref ref="kafka"/>
    </root>

    <!-- Uncomment this if you want to only change Flink's logging -->
    <!--<logger name="org.apache.flink" level="INFO">-->
    <!--<appender-ref ref="file"/>-->
    <!--</logger>-->

    <!-- The following lines keep the log level of common libraries/connectors on
         log level INFO. The root logger does not override this. You have to manually
         change the log levels here. -->
    <logger name="akka" level="INFO">
        <appender-ref ref="file"/>
        <appender-ref ref="kafka"/>
    </logger>
    <logger name="org.apache.kafka" level="INFO">
        <appender-ref ref="file"/>
        <appender-ref ref="kafka"/>
    </logger>
    <logger name="org.apache.hadoop" level="INFO">
        <appender-ref ref="file"/>
        <appender-ref ref="kafka"/>
    </logger>
    <logger name="org.apache.zookeeper" level="INFO">
        <appender-ref ref="file"/>
        <appender-ref ref="kafka"/>
    </logger>

    <!-- Suppress the irrelevant (wrong) warnings from the Netty channel handler -->
    <logger name="org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline" level="ERROR">
        <appender-ref ref="file"/>
        <appender-ref ref="kafka"/>
    </logger>
</configuration>

flink run -m yarn-cluster -yd -ytm 2g -ys 4 -yjm 2g -ynm flink-demo测试 \
-c com.eights.carbond.streaming.sensor.AverageSensorReadings \
./flink-demo-1.0-SNAPSHOT.jar