두번째 과제 진행 : By Python Code

1. CUBRID CDC 기반 Kafka Producer 구현

test_dmy_table의 변경사항을 감지해 Kafka 토픽으로 발행하는 producer를 Python으로 직접 구현함

1.1 배경

기존 HDFS 적재 메인 루프와 동일한 CDC(Change Data Capture) 패턴을 그대로 재사용하되, 목적지만 Kafka로 교체하는 작업이다.

• 문제 — 메인 루프(HDFS)와 상태 컬럼을 공유하면 두 프로세스가 서로 행을 뺏어가는 충돌이 발생함
• 해결 — test_trigger_table과는 완전히 별개인 test_kafka_trigger_table을 새로 만들고, test_dmy_table에 INSERT/UPDATE 트리거 2개를 추가로 걸어 독립적으로 이벤트를 소비하도록 구성함

---

2. 스키마 구성

CREATE TABLE NCSC_INTEG.test_kafka_trigger_table (
    trig_id          INT             AUTO_INCREMENT,
    tbl_nm           VARCHAR(100)    NOT NULL,
    pk_vl            VARCHAR(200)    NOT NULL,
    trig_type_cd     CHAR(1)         NOT NULL,
    chg_dt           DATETIME        NOT NULL,
    job_id           VARCHAR(100),
    job_proc_dt      DATETIME,
    job_prcs_stts_cd CHAR(1)         DEFAULT '0',
    retry_cnt        INT             DEFAULT 0,
    PRIMARY KEY (trig_id)
);

CREATE TRIGGER test_kafka_ins_trigger
AFTER INSERT ON NCSC_INTEG.test_dmy_table
EXECUTE INSERT INTO NCSC_INTEG.test_kafka_trigger_table (tbl_nm, pk_vl, trig_type_cd, chg_dt)
VALUES ('test_dmy_table', obj.mng_no, 'i', SYSDATETIME)
COMMENT 'kafka producer trigger';

CREATE TRIGGER test_kafka_upt_trigger
AFTER UPDATE ON NCSC_INTEG.test_dmy_table
EXECUTE INSERT INTO NCSC_INTEG.test_kafka_trigger_table (tbl_nm, pk_vl, trig_type_cd, chg_dt)
VALUES ('test_dmy_table', obj.mng_no, 'u', SYSDATETIME)
COMMENT 'kafka producer trigger';

2.1 접속 정보

• CUBRID — pycubrid.connect(host='10.189.128.210', port=33000, database='gcloud', user='dba', password='Dba123#'), dba 계정 사용이라 테이블 참조 시 NCSC_INTEG. 스키마 접두사 필요함
• Kafka — Producer({'bootstrap.servers': 'vp-bigdata-dev01:9092'}), 실제로는 dev01~03 3노드 클러스터이나 클라이언트가 메타데이터로 나머지 노드를 자동 인식하므로 dev01만 지정해도 됨
• 토픽 — test_dummy_topic, Kafbat UI로 생성 (Partitions=1, Replication factor=1, Min ISR=1 — 단일/소규모 클러스터 기준)

---

3. 스크립트 구조

단계별로 직접 작성한 구조이다.

단계	내용
1	while True: + time.sleep(10) 뼈대 구성
2	conn = pycubrid.connect(...), producer = Producer(...)를 루프 밖에서 초기화
3	클레임 UPDATE — job_prcs_stts_cd 0→1, job_id에 uuid.uuid4() 부여
4	JOIN SELECT — test_kafka_trigger_table + test_dmy_table, pk_vl=mng_no, WHERE job_id=? AND job_prcs_stts_cd='1' → rows = cur.fetchall(), columns = [d[0] for d in cur.description]
5	for row in rows: 반복하며 dict(zip(columns, row)) → json.dumps → producer.produce(topic, value=json_str, callback=...)
6	성공/실패 UPDATE — job_prcs_stts_cd 2 또는 9 + retry_cnt+1, job_proc_dt=SYSDATETIME + conn.commit()

---

4. 트러블슈팅

구현 과정에서 겪은 버그 5건을 정리하면 다음과 같다.

1) NameError: name 'json' is not defined

• 원인 — 파일 상단에 import json 누락
• 해결 — import 문 추가

2) TypeError: object of type Decimal is not JSON serializable

• 원인 — CUBRID의 숫자 컬럼이 파이썬 Decimal 타입으로 반환되는데, json.dumps()가 기본으로 처리하지 못함
• 해결 — json.dumps(row_dict, default=str)로 변경

3) 상태가 영원히 '1'에서 안 바뀌는 좀비 행 발생

• 원인 — 성공/실패 UPDATE 쿼리 뒤에 conn.commit()을 빠뜨림. 파이썬 자기 세션 안에서는 반영된 것처럼 보여도 DBeaver 등 다른 세션에는 반영되지 않음
• 해결 — commit() 추가. 이미 멈춰있던 행은 UPDATE ... SET job_prcs_stts_cd='0', job_id=NULL WHERE job_prcs_stts_cd='1'로 수동 복구 (기존에 정한 "좀비 행은 수동 대응" 정책 그대로 적용)

4) 상태는 '2'(성공)로 찍히는데 실제 Kafka에는 메시지가 없음

• 원인 — producer.flush()가 반환하는 remaining 값이 0이라는 것이 "전부 성공"을 의미하지 않음. 실패한 메시지도 처리 완료로 카운트되어 큐에서 빠지기 때문. remaining==0은 "다 끝났다"는 뜻이지 "다 성공했다"는 뜻이 아니다. 근본 원인은 애초에 rows가 0건이었던 것으로 확인됨(5번과 동일 원인) — for 루프가 한 번도 돌지 않아 produce() 자체가 호출된 적이 없는데도 errors 리스트가 비어있어 "성공"으로 오판한 것
• 해결 — producer.produce()에 delivery callback을 등록해 errors 리스트에 실제 실패만 모으고, len(rows)를 직접 출력해 확인하는 방식으로 진단

조건 0개는 항상 통과되는 것과 같은 계열의 공허한 참(vacuous truth) 버그로, 예전 NiFi RouteOnAttribute에서 겪은 문제와 동일한 패턴이다.

5) JOIN 결과(rows)가 계속 0건

• 원인 추적 — pk_vl과 mng_no 값 자체는 육안 비교/LENGTH 비교로는 동일했음. LEFT JOIN으로 전체 매칭 여부를 다시 확인한 결과, 필터 없는 순수 pk_vl=mng_no JOIN 자체는 정상 매칭됨(total_rows == matched_rows)
• 결론 — 근본 원인은 JOIN 조건이 아니라 WHERE절에서 참조하는 job_id 값이 제대로 들어가지 않았을 가능성이 높음. NiFi 버전에서 동일 패턴의 버그가 이미 확정된 바 있어, Python 버전도 파라미터 바인딩 값 또는 job_id 변수 자체의 문제로 추정되나 정확한 원인 한 줄은 재확인이 필요함

---

5. 실행 방법

pycubrid는 순수 Python 라이브러리라 가상환경 없이 시스템 python3.10에서 바로 실행 가능하다.

nohup python producer.py > producer.log 2>&1 &   # 백그라운드 실행
ps aux | grep producer.py                          # 실행 확인
tail -f producer.log                                # 로그 확인
kill <PID>                                          # 종료

---

요약

항목	내용
목표	CUBRID 트리거 기반 CDC로 Kafka 메시지 발행
핵심 설계	메인 루프(HDFS)와 독립된 트리거 테이블·트리거 분리
주요 버그	import 누락, Decimal 직렬화, commit 누락, 공허한 참 오판, JOIN 조건 미스매치
검증 결과	트리거 적재 → producer 폴링/클레임/JOIN/발행/상태 업데이트 전 과정, Kafbat UI에서 메시지 적재 확인 완료

remaining==0이 곧 전부 성공을 의미하지 않는다는 점과, for 루프가 안 돌면 실패 로그조차 안 남는다는 점 — 이 두 가지가 이번 구현에서 가장 오래 걸린 함정이었다.