쏘니의 개발블로그

Spring Boot 2.x -> 3.0 차이

• 최소 요구사항 변경 (M4 기준)
  • Gradle 7.5
  • Groovy 4.0
  • Jakarta EE 9
  • Java 17
  • Kotlin 1.6
  • Hibernate 6.1
  • Spring Framework 6
• AOT maven, gradle 플러그인 제공
• native 지원 기능 확대

spring 3.0 지원 라이브러리

• Spring AMQP 3.0.
• Spring Batch 5.0.
• Spring Data 2022.0.
• Spring GraphQL 1.1.
• Spring HATEOAS 2.0.
• Spring Integration 6.0.
• Spring Kafka 3.0.
• Spring LDAP 3.0.
• Spring REST Docs 3.0.
• Spring Retry 2.0.
• Spring Security 6.0 (see also what’s new).
• Spring Session 3.0
• Spring WS 4.0.

• Couchbase Client 3.4
• Ehcache 3.10
• Elasticsearch Client 8.5
• Flyway 9
• Groovy 4.0
• Hibernate 6.1
• Hibernate Validator 8.0
• Jackson 2.14
• Jersey 3.1
• Jetty 11
• jOOQ 3.16
• Kotlin 1.7.20
• Liquibase 4.13
• Lettuce 6.2
• Log4j 2.18
• Logback 1.4
• Micrometer 1.10
• Micrometer Tracing 1.0
• Neo4j Java Driver 5.2
• Netty 4.1.77.Final
• OkHttp 4.10
• R2DBC 1.0
• Reactor 2022.0
• SLF4J 2.0
• SnakeYAML 1.32
• Tomcat 10
• Thymeleaf 3.1.0.M2
• Undertow 2.2.20.Final

Spring의 AOT란? (Ahead Of Time)

Spring AOT 엔진은 빌드 시 스프링 애플리케이션을 분석하고 최적화하는 도구입니다. 또한 AOT 엔진은 GraalVM Native Configuration이 필요로 하는 reflection configuration을 생성해줍니다. 이것은 Spring native 실행 파일로 컴파일 하는데 사용되고 이후에 애플리케이션의 시작 시간과 메모리 사용량을 줄일 수 있게 됩니다.

Spring Boot 3.0 AOT

Spring Boot 3.0 AOT 부분 확대

위 그림에서 보면 AOT가 Spring Boot 환경에서 하는 일들과 순서를 알 수 있습니다. 간단하게 얘기하자면 Bytecode를 분석하고 최적화해서 좀 더 실행하기에 빠르고 메모리적으로 효율적인 코드를 만듭니다.

(+ spring의 native-image는 JVM에서 실행되는 파일에 비해 빌드 시간은 길고 시작시간이 짧고 메모리는 적게 사용하게 된다.)

AOT 적용 효과

• 런타임시 Spring 인프라를 적게 사용
• 런타임 시 검증할 조건 수 감소
• 리플렉션을 줄이고 프로그래밍적 Bean 등록 방식 사용

제일 메이저한 변화는 java17 이다. 또한 R2DBC 지원도 눈에 띈다.

참고문헌

https://www.baeldung.com/spring-boot-3-spring-6-new

https://github.com/spring-projects/spring-boot/wiki/Spring-Boot-3.0-Release-Notes  - github release note (자세한 버젼 정보)

카프카 오류

Synchronous auto-commit of offsets {topic-2=OffsetAndMetadata{offset=96, leaderEpoch=0, metadata=''}} failed: Offset commit cannot be completed since the consumer is not part of an active group for auto partition assignment; it is likely that the consumer was kicked out of the group.

위와 같은 오류가 발생해서 찾아보니 다음과 같다.


Kafka에서 CommitFailedException이 나오는 유형은 아래와 같다.

consumer 로직의 처리시간이 max.poll.interval.ms보다 클 경우 리밸런싱으로 인한 컨슈머 그룹에서 제외되었을 경우

session.timeout.ms시간동안 heartbeat가 오지 않았을 경우 리밸런싱으로 인한 컨슈머 그룹에서 제외되었을 경우

1. consumer 로직의 처리시간이 max.poll.interval.ms보다 클 경우 리밸런싱으로 인한 컨슈머 그룹에서 제외되었을 경우

ConsumerConfig 설정 방법 
아래 방식대로 consume 로직 처리시간 변경가능.

@Bean
public ConsumerFactory<String, JsonSerializable> consumerFactory() {
	...
    props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 5000); // 5초로 설정
    ...
}

위의 로직이 완료되고 commit될 때 CommitFailedException이 발생한다. 

max.poll.interval.ms 는 기본 5분으로 알고 있다.

This error handler cannot process 'org.apache.kafka.clients.consumer.CommitFailedException's; no record information is available

2. session.timeout.ms시간동안 heartbean가 오지 않았을 경우 리밸런싱으로 인한 컨슈머 그룹에서 제외되었을 경우

heartbeat 메시지는 consumer에서 3초(기본값)에 한번씩 주기적으로 broker에 날린다.
메시지 내용을 확인하려면 logger에 아래 내용을 추가한다.

logging:
	level:
    	org.apache.kafka.clients.consumer.internals.AbstractCoordinator: debug

subscriber의 로직에 breakpoint를 걸면 heartbeat 메시지가 날라가지 않으므로 10초 후에 리밸런싱이 되고 컨슈머 그룹에서 제외된다.
그때 로직이 수행되고 커밋하려면 CommitFailedException이 발생한다.

https://logz.io/blog/logstash-grok/

https://www.elastic.co/kr/blog/a-practical-introduction-to-logstash

grok debugger

https://grokdebug.herokuapp.com/

도커를 실무에서 사용하고 있어서 책을 통해서 더 배울게 있나 했는데, 도커교과서 라는 교재를 보니 아직 알아야 할게 너무 많은 걸 알았다.

요즘 도커를 안쓰고 개발을 하기가 어렵다. 실무에서 쓰지 않는다고 하더라도 도커는 알아야 외부 개발자와의 간단한 미팅에서라도 무시당하지 않는다. 개발자 친구가 도커를 전혀 모른다고 하면 내심 무시당할 수 있다. 자세히는 모르더라도 도커가 뭔지는 알아야 한다. 또한 요즘 쿠버네티스 또한 안하는 회사를 점차 찾기 힘들어졌다. 이런 쿠버네티스를 도입하려면 도커는 정말 필수적으로 알아야 하는 기본 지식이 되어버렸다.

이 책을 보면 도커를 처음 시작하는 사람부터 실습가능하게 설명이 되어 있다. 

도커의 기본적인 사용법부터 Dockerfile 로 도커이미지 생성하는 내용을 통해 기본적인 도커 방식 및 개념을 이해 할 수 있다. 

예제로 자바 및 node.js, go 코드에 대한 샘플도 있다. 

기본적인 내용을 학습하고 나면 도커 이미지를 저장할 레지스트리 및 도커 볼륨 등 옵션적인 내용도 학습할 수 있다. 물론 도커를 한다면 도커컴포즈 내용도 알아야 하기 때문에 해당 내용도 교재에 잘 정리되어 있다. 

그 이후의 내용은 도커를 사용한다고 해도 잘 모를 수 도 있는 내용들이었고, 이에 대해 학습할 수 있어서 이 교재를 더 추천하는 이유다.

컨테이너화된 애플리케이션에 사용되는 모니터링 기술 스택 및 도커 스웜을 통한 분산 애플리케이션 배포 등등 쿠버네티스를 학습하기 전 이 책을 통해 도커에 대해 정확히 알고 넘어가야 할 것 같다.

이 책을 필요로 할 사람으로는 도커를 실무에 적용하고 싶은 분 및 실무에 도커를 적용하나 간단하게 적용 하거나 잘 모르는 부분이 있어 정확히 알고자 하는 분 이 좋을 것 같다. 혹시나 이제 개발을 입문하는 학생 및 교육과정을 수료하는 취준생들에게는 맞지 않을 것 같다.

여태 도커에 대해 잘 몰랐던 거 같아 자책이 들지만, 이 책을 하루빨리 정독해서 바로 다음 스텝 쿠버네티스로 넘어가야 겠다.

구매정보

http://www.yes24.com/Product/Goods/111408749

Go 설치 방법

go 현재 버전 1.19.4입니다. 타르볼을 다운로드하기 전에 공식 이동 다운로드 페이지를 방문하여 버젼을 확인한다.

01. 파일 다운로드

Go 바이너리를 다운로드하려면 wget 또는 curl을 사용한다.

wget https://dl.google.com/go/go1.19.4.linux-amd64.tar.gz

02. 압축 풀기

이전 go 파일 삭제 후 tar 명령을 사용하여 /usr/local 디렉토리에 설치한다.

$ rm -rf /usr/local/go && tar -C /usr/local -xzf go1.19.4.linux-amd64.tar.gz

해당 경로에 go 폴더가 생긴다. 

$ cd /usr/local/go/bin
$ ./go version
go version go1.19.4 linux/amd64

go 실행을 환경변수에 지정하기

go 실행을 아무 경로에서 자유롭게 실행하기 위해서 ~/.bash_profile 에 경로와 GOPATH를 지정합니다.

$ vi ~/.bash_profile
export GOPATH=$HOME/go
export PATH=$PATH:$GOPATH/bin
export PATH=$PATH:/usr/local/go/bin

지정한 것만으로는 바로 반영이 아니되므로, 재로그인 아니면

$ source ~/.bash_profile

파일을 저장하고 다음 명령을 사용하여 새 PATH 환경 변수를 현재 셸 세션에 로드합니다.

source ~/.bash_profile

공식문서

https://go.dev/doc/install

circuit breaker 가 오픈되었는지 닫혀있는지를 확인하려면 대시보드가 필요하다는 생각이 든다.

예전에 histrix 를 이용할때는 turbin 이라는 모니터링이 있던거로 기억하는데, spring cloud gateway 의 circuit breaker 인 

Resilience4j 는 없다.

Resilience4j 에는 모니터링 툴이 존재 하지 않아 micormeter 로 metric 제공하는 내용으로 모니터링을 직접 구성해야 한다.

메트릭을 수집하고 표현하는 대시보드는 prometheus 와 grafana 로 모니터링을 할 수 있다.

준비 

1. Prometheus 설치

2. grafana 설치

스프링 yml

cloud:
  gateway:
    routes:
      - id: test
        uri: http://localhost:9091
        predicates:
          - Path=/circuit/**
        filters:
          - name: Retry
            args:
              retries: 3
              method: GET
              backoff:
                firstBackoff: 50ms
                maxBackoff: 500ms
          - name: CircuitBreaker
            args:
              name: myCircuitBreaker #서킷브레이커 명.
              fallbackUri: forward:/fallback
              statusCodes:
                - 500

예시로, 요청하려는 서버의 응답코드가 500 이면 서킷브레이커가 가동되고, 서킷브레이커의 네이밍을 지정할 수 있다. 이 네이밍 myCircuitBreaker 는 dashboard 에 찾을 수 있다.

스프링 

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-circuitbreaker-reactor-resilience4j</artifactId>
</dependency>
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
</dependency> <!-- 모니터링 -->

모니터링을 위해 필요한 metric 을 제공하는 라이브러리는 micrometer-registry-prometheus 이다.

그리고 actuator 가 있어야, prometheus 가 데이터를 수집해줄 수 있다.

actuator 엔드포인트 /actuator/prometheus 로 데이터 수집한다.

http://localhost:포트/actuator/prometheus

과 같이 나온다.

프로메테우스 구성법

https://juntcom.tistory.com/188

prometheus.yml 

서버에서 실행시

vi /etc/prometheus/prometheus.yml

global:
  scrape_interval: 10s # 10초마다 매트릭을 수집 default 1분
  evaluation_interval: 1m # 1분마다 규칙을 평가 default 1분

  external_labels: # 외부 시스템에 표시할 이 서버의 레이블
    monitor: 'monitor'

rule_files: # 규칙을 로딩하고 evaluation_interval 설정에 따라 정기적으로 평가한다.
# - "first.rules"
# - "second.rules"

scrape_configs:
  - job_name: 'circuit-breaker' 
    metrics_path: '/actuator/prometheus' # 메트릭을 수집할 path 설정
    static_configs:
      - targets: [‘ip:포트’] #로컬도커인 경우 'host.docker.internal:포트'

도커 실행
docker run -d -p 9090:9090 -v /etc/prometheus:/etc/prometheus --name prometheus prom/prometheus

프로메테우스 실행 하면 
http://localhost:9090/targets  에 수집하고 있는 target 들 현황을 볼 수 있다.

위의 데이터 수집까지 되었으면 prometheus 로 수집한 데이터를 grafana 로 데이터 보여주면 된다.

grafana 설치까지 하면 인프라 구성은 완료 

grafana 설치했으면 prometheus 를 datasource 추가하자.

다음에

https://resilience4j.readme.io/docs/grafana-1 대시보드 설치해주면된다.

https://github.com/resilience4j/resilience4j/blob/master/grafana_dashboard.json 

위 json 파일을 대시보드로 import 해주면 끝.

혹 데이터가 보이지 않으면, datasource 를 방금 추가해준 prometheus 설정으로 수정 하면 된다.

acks 란

acks는 acknowledgments의 약자로 사전에서 찾아 보면 "승인", 확인
프로듀서가 메시지를 보내고 그 메시지를 카프카가 잘 받았는지 확인을 할 것인지 또는 확인을 하지 않을 것인지를 결정하는 옵션

acks 옵션

acks=0 

단순 메트릭 정보와 같이 메세지 손실이 어느정도 눈감아지는 상황인 경우 사용.

acks=1

producer 가 kafka 에 데이터 전송 -> leader broker 는 쓰기요청에 대해 producer 에게 respond 를 보낸다. 그리고 데이터를 topic 에 쓴다.

leader 에게 응답받지 못한 producer 는 다시 쓰기 요청을 재시도 하고, leader 의 replica 들도 제대로 메시지를 받았는지 확인할 수 없다.

replica 들이 제대로 leader 의 topic/partition 을 제대로 복제 못한 상태에서 leader 가 다운되면 메시지가 손실 될 수 있다.

acks = all

leader 뿐 아니라 replicas 들도 ack 를 보내야 한다.

그만큼 지연시간은 늘어나지만 데이터 손실이 되지 않는다. 하나의 데이터도 손실이 용납되지 않는 경우 해당 옵션 사용 해야 한다.

acks가 all 인 경우 min.insync.replicas 옵션을 고려해야 한다.

min.insync.replicas=2 는 적어도 2개의 브로커가 데이터를 받았다는 ack응답을 보내야 한다는 뜻.

replication factor=3, min.insync.replicas=2, acks=all 인 경우 -> 모든 브로커에게 ack를 받지 못해도 min.insync.replicas 가 2이기때문에 1개가 장애나도 프로세스는 유지된다. 2개가 문제인 경우 producer 는 NOT_ENOUGH_REPLICAS 라는 예외를 받는다.

retry

NOT_ENOUGH_REPLICAS 과 같은 일시적 장애가 생긴경우 다시 시도하는 횟수.

kafka 2.1 버젼 이상부터는 retries 가 2147483647 번으로 기본값 설정됨.

retry.backoff.ms 옵션의 기본값은 100ms 이다.

delivery.timeout.ms

위 옵션이 2분인 경우 producer 는 acks 를 받지 못한 메시지에 대해 2분동안 요청 retry 를 한다.

위 시간동안 ack 를 받지 못하면 fail 이다.

retries 가 일어나는 동안 메시지의 순서는 보장되지 않는다.

key 기반 ordering 을 원하는 경우 문제가 발생한다.

이 경우 product request 가 병렬적으로 실행되도록 조절하는 옵션인 

max.in.flight.requests.per.connection 의 값을 조절하면 된다.

max.in.flight.requests.per.connection 

Default: 5

순서를 보장하기 위해서는 1로 설정해야하지만, 처리량은 낮아질 수 있다.

acks 를 보내는 과정에서 네트워크 에러가 발생하는 경우 request 가 중복이 되는 경우가 발생할수 있다.

kafka 0.11 버젼부터는 Idempotent request 라는 설정이 가능하다.

produce request 에 id 가 부여되기 때문에 broker 단에서 중복되는 reqeust 를 알아채고 중복을 방지한다.

Batching

max.in.flight.requests.per.connection = 5 의 경우 동시에 5건의 메시지가 개별적으로 전달된다. 

linger.ms -> produce 가 batch 를 보내기 전에 기다리는 시간 (ms)

값을 올릴수록 해당 초까지 기다렸다가 같이 보낸다.하지만 해당 시간 전에 batch.size 만큼 메시지가 차면 바로 배치로 보낸다.

batch.size 

하나의 배치안 에 넣을 수 있는 최대 바이트 수.

배치 사이즈를 늘리면 요청을 보낼때 압축률, 처리량, 효율성에 이점을 볼 수 있다. 

배치 사이즈보다 더 큰 사이즈의 메시지의 경우 배치로 처리되지 못한다.

하나의 배치는 파티션별로 할당되기 때문에 너무 높은 값으로 정하면 메모리 부족이 생길 수 있다.

memory

broker가 메시지를 처리하는 속도보다 producer 가 더 빠르게 메시지를 보내는 경우 해당 레코드는 잠시 producer 메모리에 buffer 된다.

buffer.memory 

기본값은 32MB, send buffer 의 사이즈다.

이 버퍼는 시간이 지남에 따라 계속 차고, 브로커에 더 빠르게 메시지를 보낼 수 있게 되면 다시 내려간다.

버퍼가 가득 차면 send 메소드는 block 된다. 데이터를 못보내고 그냥 대기하게 된다.

max.block.ms

send() 메소드가 예외를 던지기까지 block 되는 시간이다.

예외를 던지는 경우 

1. producer 버퍼 메모리가 꽉 찬 경우.

2. broker 가 전혀 새로운 데이터를 받지 못한 경우.

3. max.block.ms 시간이 다 지난 경우.

https://www.popit.kr/kafka-%EC%9A%B4%EC%98%81%EC%9E%90%EA%B0%80-%EB%A7%90%ED%95%98%EB%8A%94-producer-acks/

https://4betterme.tistory.com/168

logstasth 로 elasticsearch 에 데이터를 넣어줄 수 있다.

나는 kafka 데이터를 logstash 로 컨슘하고 output 저장소로 elasticsearch 에 넣으려고 한다.

https://www.elastic.co/guide/en/logstash/6.6/installing-logstash.html

이거대로만 하면 logstash 설치 가능하다.

요약하자면

centos 기준, elasticsearch version 6 기준

1. 퍼블릭키를 다운

rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

2. 경로 /etc/yum.repos.d/ 에서 logstash.repo 라는 파일 생성 후 아래 코드 작성

[logstash-6.x]
name=Elastic repository for 6.x packages
baseurl=https://artifacts.elastic.co/packages/6.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md

3. 설치

sudo yum install logstash

conf 파일 작성.

/etc/logstash/conf.d/ 경로에 logstasth.conf 파일 생성 후 

내용 작성한다.

input 및 output 설정해주면 된다. -> Pipline.yml 에 

path.config: "/etc/logstash/conf.d/*.conf"

과 같이 파이프라인 파일에 conf 파일을 읽는 거로 설정되어 있기 때문에 저 경로에 파일 생성해준다.

input {
    kafka {
        bootstrap_servers =>  "172.11.11.11:9092"
        group_id => "logstash"
        topics => [“test-topic”]
        consumer_threads => 1
    }
}

filter {
}

output {
    stdout {
        codec => rubydebug
    }    
    elasticsearch {
        hosts => "172.11.22.22:9200"
        index => "kafka-test-%{+YYYY-MM-dd}"
        document_type => "_doc"
    }
}

로그스태시 실행(start, restart, stop)

 systemctl restart logstash

- 로그 스테이스 설정 파일의 변경 사항을 적용하려면 재시작을 한다.

로그스테이시 로그

$ tail -f /var/log/logstash/logstash-plain.log

https://medium.com/geekculture/data-pipeline-from-kafka-to-elastic-search-using-logstash-5edca8d44d82