git clone https://gitee.com/dengmin/logback-kafka-appender.git
cd logback-kafka-appender
mvn clean install

<dependency>
    <groupId>com.gitee.dengmin</groupId>
    <artifactId>logback-kafka-appender</artifactId>
    <version>1.0</version>
</dependency>

<springProperty scope="context" name="name" source="spring.application.name"/>

# Замена темы и хостов в аппендере
<appender name="kafka" class="com.gitee.dengmin.logback.KafkaAppender">
    <appName>${name}</appName>
    <topic>cloud_logs</topic>
    <hosts>10.10.47.6:9092,10.10.47.9:9092</hosts>
</appender>
<appender name="kafka-async" class="ch.qos.logback.classic.AsyncAppender">
    <queueSize>4096</queueSize>
    <includeCallerData>true</includeCallerData>
    <appender-ref ref="kafka"/>
</appender>
```## Конфигурация интерцептора

```java
@Override
public boolean preHandle(HttpServletRequest request,
                          HttpServletResponse response,
                          Object handler) throws Exception {
    String REQUEST_ID = "REQUEST-ID";
    MDC.put(REQUEST_ID, UUID.randomUUID().toString().replaceAll("-", ""));
    return true;
}

@Override
public void afterCompletion(HttpServletRequest request,
                             HttpServletResponse response,
                             Object handler, Exception ex) throws Exception {
    MDC.clear();
}
}

@Configuration
public class AutoConfiguration implements WebMvcConfigurer {
    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(new LogMDCInterceptor()).addPathPatterns("/**");
    }
}

input {
    kafka {
        bootstrap_servers => "10.10.47.6:9092,10.10.47.9:9092"
        topics => ["cloud_logs"]   # тема, указанная в аппендере
        consumer_threads => 5
        decorate_events => true
        codec => json {charset => "UTF-8"}
        auto_offset_reset => "latest"
        client_id => "kafka_log_collect"
    }
}

output {
    elasticsearch {
        hosts => "10.10.47.10:9200"
        # Для старых версий Logstash и новых версий формат отличается, в новой версии требуется добавление [kafka]
        index => "%{[@metadata][topic]}-%{+YYYY-MM-dd}"
        index => "%{[@metadata][kafka][topic]}-%{+YYYY-MM-dd}"
    }
    # stdout {codec=>rubydebug}
}
```### Формат данных, отправляемых в Kafka

### Создание маппинга в Elasticsearch