CODEWITHSUNDEEP
javaapache-kafkalogbackspring-kafka
Hua
Hua

Reputation: 746

Redirecting kafka log messages into a separate log (other than catalina.out)

I have a project set up using Spring Boot, where I am using Spring Kafka to consume messages. The application is deployed on an independent Tomcat instance. Spring Kafka is generating a lot of logging messages, and they automatically go to catalina.out. Is there a way for me to redirect these Kafka logging messages into a separate log that I have created for my application (dataRiver log)?

I am using logback for logging.

This is what my logback-spring.xml looks like:

<property name="LOGS" value="${catalina.base}/logs" />

<include resource="org/springframework/boot/logging/logback/base.xml" />

<appender name="DATA-RIVER" class="ch.qos.logback.core.rolling.RollingFileAppender">
    <file>${LOGS}/DataRiver.log</file>
    <append>true</append>
    <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
        <Pattern>%p %d %C{1.} [%t-[%X{threadid}]] %m%n</Pattern>
    </encoder>

    <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
        <fileNamePattern>${LOGS}/archived/DataRiver-%d{yyyy-MM-dd}.log</fileNamePattern>
    </rollingPolicy>
</appender>

<appender name="ERROR" class="ch.qos.logback.core.rolling.RollingFileAppender">
    <file>${LOGS}/DataRiver-Err.log</file>
    <append>true</append>
    <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
        <Pattern>%p %d{ISO8601} [%t-[%X{threadid}]] - %m%n</Pattern>
    </encoder>

    <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
        <fileNamePattern>${LOGS}/archived/DataRiver-Err-%d{yyyy-MM-dd}.log</fileNamePattern>
    </rollingPolicy>
</appender>

<logger name="dataRiver" level="INFO" additivity="false">
    <appender-ref ref="DATA-RIVER"/>
</logger>

<logger name="error" level="WARN" additivity="false">
    <appender-ref ref="ERROR"/>
</logger>

And here's my logging service:

public class LoggingService {

  public static final Logger LOGGER_DATA_RIVER = LoggerFactory.getLogger("dataRiver");
  public static final Logger LOGGER_ERROR = LoggerFactory.getLogger("error");

}

Here's my consumer config:

@Bean
public ConsumerFactory<String, GenericData.Record> testConsumerFactoryFirst() {
  Map<String, Object> dataRiverProps = setTestDataRiverProps();
  dataRiverProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, env.getProperty("test1.bootstrap.servers"));
  dataRiverProps.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, env.getProperty("test1.schema.registry.url"));
  return new DefaultKafkaConsumerFactory<>(dataRiverProps);
}

private ConcurrentKafkaListenerContainerFactory<String, GenericData.Record> testKafkaListenerContainerFactory(ConsumerFactory<String, GenericData.Record> consumerFactory) {
  ConcurrentKafkaListenerContainerFactory<String, GenericData.Record> factory = new ConcurrentKafkaListenerContainerFactory<>();
  factory.setConsumerFactory(consumerFactory);
  factory.setBatchListener(true);
  return factory;
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, GenericData.Record> testKafkaListenerContainerFactoryFirst() {
  return testKafkaListenerContainerFactory(testConsumerFactoryFirst());
}

Here's the consumer:

@KafkaListener(topics = "#{'${test.kafka.topics}'.split(',')}", containerFactory = "testKafkaListenerContainerFactoryFirst")
public void consumeAvroFirst(List<Message<GenericData.Record>> list) {
  consumeJsonMessageBatch(convertAvroToJsonBatch(list), ""Kafka Consumer test-1");
}

private List<String> convertAvroToJsonBatch(List<Message<GenericData.Record>> list) {
  return list.stream().map(record -> record.getPayload().toString()).collect(Collectors.toList());
}

And here is an excerpt from catalina.out:

2019-04-20 02:16:54.190  WARN 18286 --- [ntainer#1-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-12, groupId=DataRiver1] 10 partitions have leader brokers without a matching listener, including [test.autoEvents-8, test.autoEvents-2, test.autoEvents-4, test.loginEvaluationEvents-17, test.loginEvaluationEvents-11, test.loginEvaluationEvents-5, test.loginEvaluationEvents-13, test.loginEvaluationEvents-7, test.loginEvaluationEvents-1, test.loginEvaluationEvents-19]
2019-04-20 02:16:54.320  WARN 18286 --- [ntainer#1-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-12, groupId=DataRiver1] 10 partitions have leader brokers without a matching listener, including [test.autoEvents-8, test.autoEvents-2, test.autoEvents-4, test.loginEvaluationEvents-17, test.loginEvaluationEvents-11, test.loginEvaluationEvents-5, test.loginEvaluationEvents-13, test.loginEvaluationEvents-7, test.loginEvaluationEvents-1, test.loginEvaluationEvents-19]
2019-04-20 02:16:54.320  WARN 18286 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-10, groupId=DataRiver1] 10 partitions have leader brokers without a matching listener, including [test.autoEvents-8, test.autoEvents-2, test.autoEvents-4, test.loginEvaluationEvents-17, test.loginEvaluationEvents-11, test.loginEvaluationEvents-5, test.loginEvaluationEvents-13, test.loginEvaluationEvents-7, test.loginEvaluationEvents-1, test.loginEvaluationEvents-19]
2019-04-20 02:16:54.346  WARN 18286 --- [ntainer#2-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-2, groupId=DataRiver1] 10 partitions have leader brokers without a matching listener, including [test.autoEvents-8, test.autoEvents-2, test.autoEvents-4, test.loginEvaluationEvents-17, test.loginEvaluationEvents-11, test.loginEvaluationEvents-5, test.loginEvaluationEvents-13, test.loginEvaluationEvents-7, test.loginEvaluationEvents-1, test.loginEvaluationEvents-19]

Thanks for your help!

Upvotes: 0

Views: 2157

Answers (1)

Ryan Dake
Ryan Dake

Reputation: 84

Create a new appender in logback-spring.xml. In the appender add a Filter. Create a filter class to match on the Kafka logs. Have the appender output to your desired location.

Appender:

<appender name="foo" class="com.foo.bar.SomeClass">
  <target>bar</target>
  <filter class="com.foo.KafkaFilter" />
   ...
</appender>

Filter class:

public class KafkaFilter extends AbstractMatcherFilter {
    @Override
    public FilterReply decide(Object o) {

        LoggingEvent loggingEvent = (LoggingEvent) o;

        if(loggingEvent.getMessage().contains("org.apache.kafka")) {
            return FilterReply.ACCEPT;
        }
    }
}

I recommend taking a look here at this answer: https://stackoverflow.com/a/5653532/10368579

The LogBack documentation has more information on matchers here: https://logback.qos.ch/manual/filters.html#matcher

Upvotes: 3

Related Questions