Перенаправление сообщений журнала kafka в отдельный журнал (кроме catalina.out) - PullRequest
Новая
       20

Перенаправление сообщений журнала kafka в отдельный журнал (кроме catalina.out)

0 голосов
/ 22 апреля 2019

У меня есть проект, настроенный с использованием Spring Boot, где я использую Spring Kafka для приема сообщений.Приложение развернуто на независимом экземпляре Tomcat.Spring Kafka генерирует много сообщений, и они автоматически отправляются в catalina.out.Могу ли я перенаправить эти сообщения журнала Kafka в отдельный журнал, который я создал для своего приложения (журнал dataRiver)?

Я использую logback для регистрации.

Вот как выглядит мой logback-spring.xml: 

<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>

А вот мой сервис регистрации: 

public class LoggingService {

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

}

Вот мой конфиг потребителя: 

@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());
}

Вот потребитель: 

@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());
}

А вот выдержка из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]

Спасибо за помощь!

1 Ответ

2 голосов
/ 22 апреля 2019

Создайте новый appender в logback-spring.xml. В appender добавьте фильтр. Создайте класс фильтра для соответствия в журналах Kafka. Получите вывод приложения appender в желаемое место.

Appender: 

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

Класс фильтра: 

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;
        }
    }
}

Я рекомендую взглянуть здесь на этот ответ: https://stackoverflow.com/a/5653532/10368579

Документация LogBack содержит больше информации о сопоставителях здесь: https://logback.qos.ch/manual/filters.html#matcher

...