이벤트 드리븐 아키텍처에 CDC 적용 수기

TL;DR

• 마이크로서비스 간의 잦은 통신으로 강결합 상태가 일어난다면 Materialized view를 사용하여 의존성을 제거해보자.

• 데이터 동기화를 위해 Event sourcing을 사용하였다면 CDC를 염두해보자.

• 이벤트 드리븐 아키텍처에 CDC를 처음 도입한다면 Outbox Pattern 을 적극 사용해보자

시작은 어땠나?

마이크로서비스와 이벤트 드리븐를 사용하다보면 고질적으로 대두되는 서비스간의 의존성 이슈가 있다. 각 서비스의 독립성을 유지하는 것이 핵심인 MSA에서 자칫하면 서비스간의 잦은 요청이 발생하는 강결합 상태가 될 수 있는 것이다.

이에 대한 대안으로, 서비스가 필요하다면 Projection 성격의 Materialized view를 생성하는 것이었다. 각 Service Boundary에서 변경이 생길 경우 Event를 발행하고 각 consumer 서비스는 Materialized View에 맞게 데이터를 동기화 하는 것이다.

출처 & 참고
The dark side of events eda

계획

나 역시 서비스간의 강결합 부분이 있다고 판단, 이를 끊어내며 CDC를 사용하기로 마음 먹었다. CDC 도 몇가지 솔루션이 존재하지만 Mysql을 사용한다면 Debezium은 매우 좋은 선택이 될 수 있다.

Jdbc source connector vs Debezium Mysql Connector
굳이 Debezium이 cdc 솔루션의 강자로 떠오르는 이유는 아래와 같다.

Kafka Connect JDBC vs Debezium CDC
jdbc source connector는 증분 쿼리 방식이다. 그래서 삭제 감지를 하지 못한다. 반면 Debezium은 mysql의 binlog를 기반으로 레코드를 파악하기 때문에 데이터(row)의 변경사항뿐만아니라 테이블 스키마의 변경도 감지가능하다.

구현방식

Debezium에 대해 스터디를 하다보면 심심치않게
db - kafka connect(debezium) - kafka - kafka connect(jdbc sink) - db
구조가 보인다.

꼭 이렇게 쓰지 않아도 된다. 좀 더 정확히 말하면 상황에 따라 sink구간이 바뀌어야 한다. 후반 부에 이야기 하겠지만 나는 이 그림이 정석인 줄 알고 따라 했다가 큰 시행착오 겪었다.

실행

해본 것

세세하게 설명할 수는 없지만 아래 항목들을 진행했다.

• CDC 대상 데이터베이스 설정 및 Failover 정책 수립
• Kafka Connect, Kafka Registry, UI 관리툴 구성
• 부하 테스트

CLI로만 CDC를 운영하기엔 무리가 있을 것으로 판단해서 UI 관리툴을 선정하려했다.
(Akhq, UI for Apache kafka, Kakao Genesis, Debeziu UI, Kafka-Connect UI)
최종적으로 UI for Apache kafka를 선정하였는데, 자세한 내용은 추후 포스팅해봐야겠다..

데이터 동기화 지연 관련..
보통 CDC를 사용하면 순서보장을 위해 source tasks.max = 1 유지하게 된다.
실제 kafka connector는 생각보다 굉장히 빨랐다. 때문에 source구간의 병목은 크지 않았다. 동기화 지연(consumer lag)이 발생했을 시, 토픽 파티션과 sink tasks.max 및 batchSize을 조절하여 쉽게 해결 할 수 있었다.

{
    "name":"CONNECTOR_NAME",
    "connector.class": "io.aiven.connect.jdbc.JdbcSinkConnector",
    "topics": "TOPIC_LIST",
    "connection.url": "DB_CONNECTION_URL",
    "connection.user": "DB_USERNAME",
    "connection.password": "DB_PASSWORD",
    "tasks.max":"1",
    "auto.create": "true",
    "auto.evolve": "true",
    "insert.mode": "upsert",
    "delete.enabled": "true",
    "pk.mode": "record_key",
    "pk.fields": "field1,field2",
    ...생략
}

JdbcSinkConnector 요렇게 생겼다.

실패

호기롭게 도전했단 시작과 달리 CDC적용은 처참하게 실패했다.
이유는 적용 대상이 이벤트 드리븐 아키텍처였기 때문이다.

실패 as-is 아키텍처

왼쪽 Service Boundary의 Domain event와 Change event는 각각 Cache와 DB를 동기화 하고 있었다. 당연히 비동기로 이루어지는 이벤트 기반 통신은 동기화 시점에 따라 DB와 Cache의 이격이 발생하였고 이는 사용자에게 전달되는 View 데이터일 수록 크리티컬한 이슈로 발생할 수 있었다.

서비스 영향을 최소화 하기 위해 Kafka Connect를 사용하여 CDC를 구성한 것이 화근이었고, 하나의 service boundary 내 동일 대상의 여러 이벤트를 발행한 것이 이유였다.

고민

힌트를 얻은곳

고민부터 결론까지 가장 큰 영향을 준 글이다.
Debezium 공식 블로그 실린 내용으로 Domain Event와 Change Event, Event sourcing과 CDC에 대한 정의와 비교를 해주어 좋은 인사이트를 얻을 수 있다.

Distributed Data for Microservices — Event Sourcing vs. Change Data Capture

가장 크게 생각을 바꾼 계기가 바로 위 그림이다.
Source와 Sink 구현 방법에 다양한 CDC 사용방법을 나타내 주고 있다.
기존 Option1/A 방법이 많았지만 Option1/B 와 Option2 방법이 더욱 확산되고 있다는 이야기다.

위에서 이야기했듯이 정석인줄 알았던 Kafka Connect (Option1/A) 방식이 상황에 따라 다르게 구현되어야되는 것을 깨달았다. (생각해보면 너무나 당연한 이야기였다.)

CDC 본질 _ 나는 왜 Sink Kafka Connect에 집착했는가

본질은 데이터변경 캡처다 즉, 데이터 변경사항을 이벤트로 전파하는 것이다.
다시 말해 CDC는 데이터 Source 구간이 관심사며, Sink구간은 CDC를 사용한 데이터 파이프라인의 관심사다. 하지만 나는 Sink구간까지 CDC라 생각하여 이미 만들어진 아키텍처에 Sink를 덧붙이는 착오를 저지른 것이었다.

결론

CDC에 집중하자

해결방안 아키텍처

생각이 바뀌고나니 의외로 답은 쉽게 찾을 수 있었다.
"이벤트를 단일화 하자"
변경 이벤트를 Service Boundary에서 하나만 발행하도록 한다.
단, 기존 아키텍처에서 사용하던 이벤트 형태를 최대한 지키면서 변경점을 최소화하도록 한다.
그렇게되면 우리는 오로지 Sink구간의 데이터 동기화만 신경쓰면 된다.

어떻게? (outbox pattern)

그러나 기존에 사용되던 Event형태로 변경이벤트를 바꾸기는 쉽지 않다.
Debezium은 자체적으로 binlog를 읽어 메세지로 변경하는 Payload 형태가 있기 때문에다.

Debezium이 Default형태로 발행하는 이벤트 Payload

{
  "schema": {...},
  "payload": {
    "before": {  
      "id": 1004,
      "first_name": "Anne",
      "last_name": "Kretchmar",
      "email": "annek@noanswer.org"
    },
    "after": {  
      "id": 1004,
      "first_name": "Anne Marie",
      "last_name": "Kretchmar",
      "email": "annek@noanswer.org"
    },
    "source": {  
      "name": "1.9.5.Final",
      "name": "dbserver1",
      "server_id": 223344,
      "ts_sec": 1486501486,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 364,
      "row": 0,
      "snapshot": null,
      "thread": 3,
      "db": "inventory",
      "table": "customers"
    },
    "op": "u",  
    "ts_ms": 1486501486308  
  }
}

이를 해결하기 위해 Outbox pattern을 사용할 수 있다.

Service Boundary내 도메인 변경사항을 Outbox 전용 테이블에 저장한다.
Debezium은 이 Outbox 테이블의 변경사항을 파악하여 이벤트로 발행한다.
여기서 Debezium은 SMT(Single Message Transform)을 지원하는데, 이는 Outbox pattern에서 메세지를 별도로 커스텀하지 않아도 Debezium 자체가 DB에 저장된 Payload형태로 이벤트를 발행해준다.

원본이벤트

# Kafka Message key: "406c07f3-26f0-4eea-a50c-109940064b8f"
# Kafka Message Headers: ""
# Kafka Message Timestamp: 1556890294484
{
  "before": null,
  "after": {
    "id": "406c07f3-26f0-4eea-a50c-109940064b8f",
    "aggregateid": "1",
    "aggregatetype": "Order",
    "payload": "{\"id\": 1, \"lineItems\": [{\"id\": 1, \"item\": \"Debezium in Action\", \"status\": \"ENTERED\", \"quantity\": 2, \"totalPrice\": 39.98}, {\"id\": 2, \"item\": \"Debezium for Dummies\", \"status\": \"ENTERED\", \"quantity\": 1, \"totalPrice\": 29.99}], \"orderDate\": \"2019-01-31T12:13:01\", \"customerId\": 123}",
    "timestamp": 1556890294344,
    "type": "OrderCreated"
  },
  "source": {
    "version": "1.9.5.Final",
    "connector": "postgresql",
    "name": "dbserver1-bare",
    "db": "orderdb",
    "ts_usec": 1556890294448870,
    "txId": 584,
    "lsn": 24064704,
    "schema": "inventory",
    "table": "outboxevent",
    "snapshot": false,
    "last_snapshot_record": null,
    "xmin": null
  },
  "op": "c",
  "ts_ms": 1556890294484
}

SMT가 적용된 이벤트

# Kafka Topic: outbox.event.order
# Kafka Message key: "1"
# Kafka Message Headers: "id=4d47e190-0402-4048-bc2c-89dd54343cdc"
# Kafka Message Timestamp: 1556890294484
{
  "{\"id\": 1, \"lineItems\": [{\"id\": 1, \"item\": \"Debezium in Action\", \"status\": \"ENTERED\", \"quantity\": 2, \"totalPrice\": 39.98}, {\"id\": 2, \"item\": \"Debezium for Dummies\", \"status\": \"ENTERED\", \"quantity\": 1, \"totalPrice\": 29.99}], \"orderDate\": \"2019-01-31T12:13:01\", \"customerId\": 123}"
}

사실 가장 큰 힌트를 얻었다고 말한 글에서도 대놓고 처음에는 Outbox Pattern을 사용하도록 적극 권장한다.

또한 Debezium 공식 Oubox Event Router에서도 이벤트 사용간의 불일치를 방지하기 위해 사용할 수 있다고 첨언한다.

참고1

• 이벤트 Outbox table에 insert 직후 delete
• payload body Json 컨버팅 transforms.outbox.table.expand.json.payload=true
• 토픽 라우팅 route.by.field (aggregatetype default)

Outbox pattern 적용을 위 항목을 포함한 몇가지 주의사항이 존재하지만 아래 링크를 보면, 특히 Spring을 사용자라면 예제를 보고 손 쉽게 Outbox pattern을 구현할 수 있다.
Outbox Event Router
Reliable Microservices Data Exchange With the Outbox Pattern
Spring boot 버전 outbox pattern 구현 예제

참고2

@Component
public class OutboxListener {
	//생략...
    @EventListener
    public void onExportedEvent(Outboxable event) {

        OutboxEvent outboxEvent = OutboxEvent.from(event);
        repository.save(outboxEvent);
        repository.delete(outboxEvent);

    }

}

예제를 보면 outbox 패턴의 특이점으로는 outbox 테이블에 entity를 입력했다가 삭제하는 것이다. 이는 outbox table을 빈 상태로 유지하기 위함이다. 조금만 생각해보면 Debezium Mysql connector는 binlog기반으로 작동하기에 insert되는 이벤트 entity에만 관심있다는 것을 알 수 있다.

참고 3
위 예시코드에 따르면outbox 테이블의 entity는 insert 즉시 delete를 실행한다.

        repository.save(outboxEvent);
        repository.delete(outboxEvent);

Outbox pattern의 핵심은 Domain 변경사항이 outbox 이벤트와 같은 트랜잭션으로 저장된다는 것이다. 때문에 하나의 트랜잭션에서 필연적으로 outbox entity를 insert/delete 하게 되는데, 이는 JPA/hibernate를 사용할 경우 optimisticlock을 마주치게 된다.
몇 가지 아래와 같은 대안이 있다.

• 별도의 outbox delete 이벤트를 만들어 후처리하기
• 비관적 lock을 사용한 delete (성능 고려 필요)
• Navtive query를 사용한 optimistic lock 회피 (delete가 정상 작동하지 않는 경우 발생)
• 별도의 batch job/scheduler를 사용한 outbox table 삭제

마무리

이 글에는 마이크로서비스 + 이벤트 드리븐 아키텍처에 CDC를 적용할 당시 경험을 담았다.

• 데이터베이스 Master-slave 구성 및 MMM 사용여부
• 데이터베이스 용량 및 binlog 삭제 주기
• CDC 전용 인스턴스 및 권한 구성 (필요에 따라 NAT사용)

자세히 묘사하지는 않았지만 CDC를 적용하기 전에 데이터베이스와 관련된 많은 준비동작이 필요하다.
또한 CDC 적용 이후 모니터링 및 운영 노하우에 적절히 고민해볼 필요가 있다. 위에 소개한 구현 방식에 따라 관리 포인트에 대한 내용이 조금씩 차이나기 때문이다.
여느 기술스택이 그렇듯 서비스에 도입은 항상 신중해야한다. 그 것이 CDC와 같이 Infrastructure에 영향을 받는다면 DBA 및 DevOps 개발분들과 긴밀히 협업할 준비도 되어있어야한다.