LG CNS 부트캠프 학습일지 50일차

학습 내용

• Kafka Connect를 사용해서 데이터베이스 동기화하기
• Kafka를 통해 서비스 간에 통신하기

Kafka Connect를 사용해서 데이터베이스 동기화하기

슬프지만 실습을 잘 따라가지 못했다. 여러 원인이 있다. 강사님이 제공해주신 docker-compose.yml 파일을 그대로 사용하지 않았기 때문이고, 도커를 로컬에서 돌리지 않고 개인 서버에서 원격으로 돌리고 있었기 때문이다. 이런 이유 때문에 강사님이 제공해주신 docker-compose.yml 파일을 그대로 사용하지 못했다.

어떤 문제가 있었냐면, Kafka Connect 에서 MariaDB 드라이버를 인식하지 못하는 것이었다.

{
  "error_code": 400,
  "message": "Connector configuration is invalid and contains the following 1 error(s):\nCould not connect to database. No suitable driver found for jdbc:mariadb://mariadb-source:3306/appdb\nYou can also find the above list of errors at the endpoint `/connector-plugins/{connectorType}/config/validate`"
}

무엇을 실습하려는 것인지는 이해했다. Order Service 마이크로서비스가 스케일 아웃(scale out)해서 여러 개 있는 상황을 상정해보자. 어제 이야기한 것처럼 문제가 하나 발생하는데, Order Service를 통해서 주문을 만들면 그 주문이 라운드 로빈 방식으로 마이크로서비스에 나뉘어서 만들어진다. 따라서 이를 동기화할 필요가 있는데, Kafka Connect가 제공하는 소스(source) 커넥터와 싱크(sink) 커넥터를 사용해서 데이터베이스 한 쪽에서 새로운 정보가 삽입(insert)되면 그 정보를 Kafka에 전달해서 다른 쪽에서도 받아서 반영할 수 있게 한다.

개념… 은 어렵지 않지만, 직접 해보지 못했다. 과제를 제출해야하는데, 그 과제를 하면서 내가 무엇을 잘못했는지 알 수 있게 될 것으로 기대한다.

Kafka를 통해 서비스 간에 통신하기

구체적으로 말하면 Order Service을 통해 주문을 하면 그 주문 수량 만큼 Catalog Service에서 재고를 반영하는 것을 구현했다. RestTemplate이나 OpenFeign을 사용해서 Order Service가 Catalog Service 엔드포인트에 접근해서 자료를 동기화하는 방식도 있겠으나, 이번 실습에서는 Kafka를 사용했다.

이를 위해서 Order Service와 Catalog Service에 Kafka를 의존성으로 추가해주어야 했다.

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

application.yml에는 아래와 같은 설정값을 넣어주었다.

spring:
  kafka:
    bootstrap-servers: ip.address.to.kafka:9092
    consumer:
      group-id: "product-service-group"
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      auto-offset-reset: earliest

application.yml 파일을 사용하지 않고 @Configuration 어노테이션을 사용한 자바 코드의 형태로 만드는 방법도 있다. 하지만 강사님의 코드는 applicaion.yml을 사용했었다. 아래 코드가 주석처리되어 남아있는 것으로 봐서는 이런 것도 가능하다는 것을 보여주고 싶으셨던 것 같다.

@EnableKafka
@Configuration
public class KafkaConsumerConfig {
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> properties = new HashMap<>();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumerGroupId");
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        return new DefaultKafkaConsumerFactory<>(properties);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory
                = new ConcurrentKafkaListenerContainerFactory<>();
        kafkaListenerContainerFactory.setConsumerFactory(consumerFactory());

        return kafkaListenerContainerFactory;
    }
}