이벤트 버스를 위한 메시지 큐로 Kafka를 사용하자.

[JunsuLime]

Kafka 를

1. 이벤트 버스
2. 스트림 프로세서
3. 메시지 큐
4. message request-reply

의 용도로 사용하자.

이 중에서 이벤트 버스로 먼저 사용하자.

[JunsuLime]

일단 로컬환경에서 Kafka 를 실행하기위해 docker-compose 에 https://hub.docker.com/r/bitnami/kafka 를 추가하자.

• zookeeper
• kafka server

를 docker-compose 로 실행하자.

로컬환경에서 실행한다는 가정하에 필요한 요소만 변수로 빼자.

[JunsuLime]

Spring cloud event bus

이벤트 버스의 구현체를 kafka 로 사용하고, 어떻게 동작하는지 파악하자. https://docs.spring.io/spring-cloud-bus/docs/current/reference/html/

https://docs.spring.io/spring-cloud-bus/docs/current/reference/html/#customizing-the-message-broker 이를 참고해 message broker 를 customizing 할 수 있다.

properties 는 여기서 확인가능 (https://cloud.spring.io/spring-cloud-bus/reference/html/appendix.html)

Spring Cloud Bus uses Spring Cloud Stream to broadcast the messages. 클라우드 버스는 Spring Cloud Stream 을 이용하여 메시지를 broadcasting 하고 있고

rabbitmq 및 kafka 의 설정은 spring cloud stream 을 확인해 작성가능하다.

Kafka Stream 의 properties

• https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/current/reference/html/spring-cloud-stream-binder-kafka.html#_configuration_options
• https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/current/reference/html/spring-cloud-stream-binder-kafka.html#_configuration_options_2

Apache Kafka 의 properties

• https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/current/reference/html/spring-cloud-stream-binder-kafka.html#_configuration_options

[JunsuLime]

그럼 위의 문서 기반으로 Kafka 설정을 마저 해보도록 하자.

https://kafka.apache.org/26/documentation.html#acks

로컬환경에서 실행하기위해서 챙겨야할 것들

어플리케이션 서버에서

• spring.cloud.stream.kafka.binder.autoCreateTopics 설정을 true 로 해줘야한다.

Docker 카프카 설정에서

• KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE 을 true 로 해준다. (topic 자동 생성을 위해서)
• https://parkcheolu.tistory.com/196 advertised listener, listener 이런게 있군 천천히 보자

zookeeper 에서는 왜 이런 에러가 발생하는 걸까? EndOfStreamException: Unable to read additional data from client, it probably closed the socket

kafka 에서는 왜 이런 에러가 발생하는 걸까?

kafka_1           | [2021-11-03 11:09:58,760] ERROR Error while creating ephemeral at /brokers/ids/1005, node already exists and owner '72116708664803328' does not match current session '72116711532724224' (kafka.zk.KafkaZkClient$CheckedEphemeral)
kafka_1           | [2021-11-03 11:09:58,787] ERROR [KafkaServer id=1005] Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)

타이밍 이슈임.... 왜 타이밍 이슈가 발생하는지는 하다가 계속나오면 다시 이슈를 만들어보자.

[JunsuLime]

카프카 기반 이벤트 버스는 카프카를 어떻게 이용해서 동작하는지 파악해보자.

[JunsuLime]

ConfigServicePropertySourceLocator 에서 spring.cloud.config.uri 프로퍼티를 사용하고있다. 그래서

• spring.cloud.config.uri
• spring.config.import

둘다 추가를 해줘야한다.

재 갱신 시 spring.cloud.config.uri 를 참고해서 호출하는것으로 파악됨 -> bootstrap 파일도 다시 복구 시켰습니다 ㅠ_ㅠ

===============================================

[Consumer clientId=consumer-anonymous.07d30449-2268-4fb5-8203-9a8a38b0f8f0-2, groupId=anonymous.07d30449-2268-4fb5-8203-9a8a38b0f8f0] Cluster ID: RWbw3vqRR2-N1njzEkaayA

인스턴스가 올라왔을 때, 임의의 groupId, clientId 가 따짐