[Connector][Debezium] Oracle 11g + Debezium

시작

아직도 일부 기관이나 옛날 서비스는 Oracle 11g 환경을 사용하는 곳이 많고 나는 이곳에서 CDC를 적용해야만 했다.

해당 데이터베이스는 수십개의 스키마, 수백개의 테이블, 수천만건의 레코드가 담긴 Oracle 10g 이며 이 데이터베이스에서 필수 데이터만 남기고 Oracle 11g로 이관 및 CDC를 심어야했다.

19c도 아니고 왜 11g이냐? 이유는 아래와 같다

1. 타 연계 애플리케이션이 최대 11g까지 지원 <-결정적 이유
2. 해당 기관은 11g이후 사용사례가 없음

이러한 이유로 11g를 Rocky9.7에 설치했었다.

설치 난관

그냥 시작부터 적용까지 쉬운 절차가 아니었다.

1. Oracle 11g를 Rocky9.7에 설치

다행히 레퍼런스가 구글링 하면 바로 나왔다.

아래 링크로 대부분 해결하였다.

VirtualBox에 Rocky Linux 9.5 설치부터 Oracle11g 설치까지(2)

2. 부족한 자원

1테라도 아니고 약 700Gb의 자원에 OS, Oracle, TableSpaces 모두 적용해야 했다.

여기에 CDC에 꼭 필요한 아카이브로그(Archive Log)공간도 넉넉히 줘야 했기 때문이다.

참고로 아카이브로그가 꽉차버리면 서비스가 불가하는 치명적인 문제가 있다.

AS-IS또한 1테라도 안되는 공간이었지만 `SUPPLEMENTAL_LOG_DATA_ALL` 옵션이 `NO` 였기 때문에 그렇게 많이 쌓이진 않았었다.

개발 난관

1. Debezium Oracle11g 미지원

Oracle 11g는 지원하지 않습니다. 하지만 최대한 Oracle 11g와의 하위 호환성을 유지하기 위해 노력하고 있습니다. Oracle 11g와의 호환성 관련 문제점을 알려주시고, 오류가 발견될 경우 버그 수정 정보를 제공해 주시면 감사하겠습니다. - debezium -

이렇게 오피셜 문서에 "Oracle 11g가 지원되나요?"항목에 박혀있다. 

구동은 가능하지만 알려지지 않은 버그가 있을수도 있다는 소리다.

2. 운영테스트 불가

CDC데이터를 직접 Producing하는것을 테스트해야하는데 이것을 운영환경에서 할 수 없었다

데이터베이스를 바꾸는순간부터 오픈이기 때문에 개발할때는 SQL Pooling방식으로 운영에 긁어서 직접 적재시켰다

DATE 컬럼데이터를 가지고 N시간의 데이터를 모두 긁어와서 Upsert해버리기때문에 또다른 문제에 직면했다.

3. 의미없는 CDC데이터

진짜 Insert, Update된 Row는 별로없는데 N시간의 데이터를 모두 긁어와 Upsert해버리니 트랜젝션 커밋이 매우많다.

변경사항 없는 Update도 한 트랜젝션이기 때문에 이것또한 모두 캡쳐를 해버린다. 이 말은 즉 아카이브로그에 로그가 매우 많이 쌓이고 있단 소리다. 이 현상은 곧 설치난관의 2번 항목(부족한 자원) 문제와 직결된다.

 

뭐 결국 난관을 극복했고 오픈까지 했으니 다행이지만 그 과정에서 스트레스가 좀 있었다..

부족한 자원은 Cron스케줄러를 통해 아카이브로그 삭제주기를 매우 보수적으로 잡고

시스템 진단데이터도 몇시간 짜리만 남겨두고 주기적으로 삭제시켰다.

이 행위는 곧 먼 미래 장애시 데이터 복구가능시간과 원인파악하는데 문제를 일으킬 수 있지만 어쩔수 없었고 PL님께 보고도 드려서 스토리지 확장도 협의했으나 현 스토리지(SSD) 가격이슈로 승인되지 않았다. 인수인계받은 유지보수팀에게도 전달하여 숙지시켰으니 이후 문제는 유지보수가 알아서 잘 하겠지..

Oracle설정부터 연결까지

1. ARCHIVELOG 모드 확인 및 설정

아카이브로그가 켜져있는지 확인한다.

SELECT LOG_MODE FROM V$DATABASE;

1.1. `NOARCHIVELOG`일경우 

sqlplus / as sysdba

SHUTDOWN IMMEDIATE;
STARTUP MOUNT;
ALTER DATABASE ARCHIVELOG;
ALTER DATABASE OPEN;

다시 확인

2. Supplemental Logging 확인

기본적으로 변경된 컬럼값만 REDO로그에 저장되는데 이것을 변경된 행(ROW)전부를 저장하도록 설정한다.

SELECT	SUPPLEMENTAL_LOG_DATA_MIN,
        SUPPLEMENTAL_LOG_DATA_ALL
FROM V$DATABASE;

2.1. 둘중 한개이상 `NO`가 있을 경우

-- SUPPLEMENTAL_LOG_DATA_MIN -> YES
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

-- SUPPLEMENTAL_LOG_DATA_ALL -> YES
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

다시 확인

3. CDC전용 계정 생성

테스트용이라 DBA권한을 줬다. 

CREATE USER CDC IDENTIFIED BY cdc
DEFAULT TABLESPACE USERS
TEMPORARY TABLESPACE TEMP;

GRANT DBA TO CDC;

운영에서는 권한을 각각 주는것이 안전하다 

4. Debezium 커넥터

아래 내용을 RequestBody에 넣고 `/connector` POST요청하면 된다.

{
  "name": "oracle11g-cdc-connector",
  "config": {
    "connector.class": "io.debezium.connector.oracle.OracleConnector",
    "tasks.max": "1",
    "database.hostname": "{DB주소}",
    "database.port": "1521",
    "database.dbname": "{DB명}",
    "database.user": "CDC",
    "database.password": "cdc",
    "database.url": "jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST={DB주소})(PORT=1521))(CONNECT_DATA=(SID={SID})))?oracle.jdbc.timezoneAsRegion=false",
    "topic.prefix": "TEST_CDC",
    "schema.include.list": "CDC",
    "schema.history.internal.kafka.bootstrap.servers": "127.0.0.1:9092",
    "schema.history.internal.kafka.topic": "schema-changes.oracle",
    "schema.history.internal.store.only.captured.tables.ddl": "true",
    "schema.history.internal.store.only.captured.databases.ddl": "true",
    "schema.history.internal.kafka.recovery.attempts": "3",
    "schema.history.internal.kafka.recovery.poll.interval.ms": "1000",
    "schema.history.internal.skip.unparseable.ddl": "true",
    "log.mining.strategy": "online_catalog",
    "log.mining.session.max.ms": "600000",
    "log.mining.archive.log.hours": "0",
    "log.mining.batch.size.min": "1000",
    "log.mining.batch.size.default": "50000",
    "log.mining.batch.size.max": "100000",
    "log.mining.sleep.time.min.ms": "0",
    "log.mining.sleep.time.default.ms": "1000",
    "log.mining.sleep.time.max.ms": "3000",
    "log.mining.sleep.time.increment.ms": "500",
    "log.mining.transaction.retention.ms": "3600000",
    "log.mining.buffer.type": "memory",
    "log.mining.buffer.transaction.events.threshold": "10000",
    "log.mining.include.redo.sql": "false",
    "log.mining.query.filter.mode": "in",
    "log.mining.restart.connection": "true",
    "snapshot.mode": "no_data",
    "snapshot.locking.mode": "none",
    "decimal.handling.mode": "string",
    "database.tablename.case.insensitive": "true",
    "tombstones.on.delete": "false",
    "errors.max.retries": "-1",
    "errors.retry.delay.max.ms": "60000",
    "producer.compression.type": "lz4",
    "producer.linger.ms": "50",
    "producer.batch.size": "131072",
    "task.restart.backoff.ms": "30000",
    "task.restart.backoff.max.ms": "300000",
    "database.query.timeout.ms": "300000"
  }
}

5. KafkaProducing 확인

topic.prefix 에 설정된 접두사 . 스키마명 . 테이블명 으로 CDC데이터가 오는 모습..

`Insert`라 before는 null로 온다.

장애

Debezium에서 몇몇 장애가 발생했을 경우이다

예를 들어 아카이브 로그 삭제주기를 4시간 지난 데이터를 삭제하고, 장애가 4시간이상 났을경우 Offset은 이미 삭제된 SCN을 바라보고 있어 커넥터에서 오류가 발생한다.

이땐 Offset를 초기화해주는 작업이 필요하고 장애발생 SCN부터 초기화 시점의 SCN까지의 데이터는 캡쳐하지 못하여 유실된다

이 때 스냅샷을 재실행 해주면 문제가 없지만 테이블의 데이터가 너무 클경우, 레코드 수가 너무 많은 테이블일 경우 스냅샷 시간과 테이블 Lock시간이 너무 길어지기 때문에 개발자는 Debezium을 연결할때 장애에 대해서 고려를 깊게 해야한다.

이번 프로젝트의 경우에는 원천 DB에서 타 시스템 연계 애플리케이션으로 들어온 데이터를 프로시저를 통해 테이블을 생성시켜준 것을 캡쳐하기 때문에 유실될 경우 원천 DB에서 쿼리로 따오면 해결 가능하지만 내 자신이 원천DB면 머리가 아팠을 것이다.

오프셋 초기화

Kafka Connect Offset 관리는 Docker 설치 기준 `OFFSET_STORAGE_TOPIC` 옵션에 정의된 토픽에서 관리된다

나는 `my_connect_offsets` 으로 설정했고 여기로 가면 현재 오프셋이 어디 SCN을 보는지 확인할 수 있다.

Key

Value

여기서 초기화를 할거면 커넥터를 삭제하고, 아래와 같이 Key만 넣어주고 Value를 비워서 Produce Message해준다음 다시 커넥터를 생성하면 초기화된다.

또는 아래 Kafka CLI로 초기화 할 수 있다.

./kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic my_connect_offsets \
  --property "parse.key=true" \
  --property "key.separator=|"
  
  # 콘솔모드로 들어가지면
  ["oracle11g-cdc-connector",{"server":"TEST_CDC"}]|
  
  # 엔터치고 `ctrl + c`로 콘솔 나오기

SCN값으로 시간 추정

추가적으로 SCN값으로 시간을 알 수 있다.