본문으로 바로가기 본문 바로가기
WIKI 기술 지식 베이스

Debezium MySQL 커넥터

원문 보기 위키 갱신

Debezium MySQL 컴포넌트는 Debezium을 Debezium Engine으로 감싸서, Kafka나 Kafka Connect 없이도 MySQL 데이터베이스의 Change Data Capture(CDC)를 할 수 있게 해 주는 컴포넌트예요. Camel 3.0부터 제공됐고 consumer만 지원합니다. 스냅샷 모드, 스키마 히스토리, GTID, 하트비트, 증분 스냅샷까지 세밀한 옵션을 제공해요.

출처: 문서

본문

Camel 3.0부터 지원

consumer만 지원해요

Debezium MySQL 컴포넌트는 Debezium을 Debezium Engine으로 감싼 wrapper예요. Kafka나 Kafka Connect 없이도 Debezium으로 MySQL 데이터베이스의 Change Data Capture를 수행할 수 있게 해 줍니다.

참고 장애 처리에 대한 참고 사항: Debezium Embedded Engine 문서에 따르면, 엔진은 소스 오프셋을 계속 기록하고 이 오프셋을 주기적으로 영속 스토리지에 플러시해요. 그래서 애플리케이션을 재시작하거나 크래시가 나도, 엔진은 마지막으로 기록된 오프셋부터 다시 이어 갑니다. 정상 동작 중에는 다운스트림 라우트가 각 이벤트를 정확히 한 번씩 받아요. 하지만 애플리케이션이 크래시로 멈추면(우아한 종료 없이) 마지막 기록 오프셋부터 이어서, 재시작 직후에 중복 이벤트를 받을 수 있어요. 따라서 다운스트림 라우트는 이런 상황을 충분히 견디도록 만들고, 필요하면 이벤트 중복을 제거해야 합니다.

엔진이 폴링을 시작하기 전에는 consumer 헬스 체크가 설정된 초기 상태(DOWN이 기본값)를 보고해요. 커넥터 태스크가 시작되고 엔진이 폴링 단계에 들어가면 변경 이벤트가 하나도 도착하지 않았더라도 UP을 보고합니다. 이것이 초기 스냅샷이 끝났다는 뜻은 아니에요.

임베디드 엔진이 에러로 멈추면(예: 커넥터가 데이터베이스에 도달할 수 없거나 오프셋 스토어를 읽을 수 없을 때) 재시작되지 않아요: 라우트는 시작 상태로 남지만 더 이상 이벤트를 받지 못합니다. 이 장애는 consumer 예외 핸들러로 보고되고 라우트 헬스 체크를 DOWN으로 바꿔서, readiness 프로브가 감지할 수 있게 해 줘요.

Maven 사용자라면 이 컴포넌트를 쓰려면 pom.xml에 다음 의존성을 추가해야 해요:

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-debezium-mysql</artifactId>
    <version>x.x.x</version>
    <!-- use the same version as your Camel core version -->
</dependency>

URI 형식 (URI Format)

debezium-mysql:name[?options]

참고 라이선스 문제 때문에 MySQL 커넥터를 사용한다면 mysql-connector-j 의존성을 추가해야 해요. POM 파일에 다음을 추가하세요:

<dependency>
    <groupId>com.mysql</groupId>
    <artifactId>mysql-connector-j</artifactId>
    <version>8.0.15</version>
</dependency>

옵션 설정 방법 (Configuring Options)

Camel 컴포넌트는 두 단계로 나눠 설정해요:

  • 컴포넌트(component) 레벨

  • 엔드포인트(endpoint) 레벨

컴포넌트 옵션 설정 (Configuring Component Options)

컴포넌트 레벨에서는 엔드포인트들이 물려받는 일반·공유 설정을 정해요. 설정 계층 중 가장 위에 해당합니다.

예를 들어 컴포넌트에는 보안 설정, 인증용 자격 증명, 네트워크 연결 URL 같은 것들이 들어갈 수 있어요.

컴포넌트에 따라 옵션이 몇 개뿐인 경우도 있고 많은 경우도 있어요. 컴포넌트는 보통 널리 쓰는 기본값을 미리 갖고 있어서, 옵션을 몇 개만 설정하거나 아예 안 건드려도 되는 경우가 많습니다.

컴포넌트는 다음 방법으로 설정할 수 있어요:

  • Component DSL.

  • 설정 파일(application.properties, *.yaml 파일 등).

  • Java 코드에서 직접.

엔드포인트 옵션 설정 (Configuring Endpoint Options)

엔드포인트는 옵션이 많아서 설정에 시간을 더 쓰게 됩니다. 이 옵션들로 엔드포인트가 원하는 동작을 하도록 조정할 수 있어요. 옵션은 엔드포인트를 consumer(from)로 쓰는지, producer(to)로 쓰는지, 둘 다 쓰는지에 따라 나눠지기도 합니다.

엔드포인트 설정은 대개 엔드포인트 URI에 _path_와 query 파라미터로 직접 하는 경우가 많아요. Java에서 엔드포인트와 데이터 포맷을 _타입 세이프_하게 설정하려면 Endpoint DSL과 DataFormat DSL을 쓸 수도 있습니다.

옵션을 설정할 때 좋은 습관은 Property Placeholders(프로퍼티 플레이스홀더)를 쓰는 거예요.

프로퍼티 플레이스홀더가 주는 이점은 이래요:

  • URL, 포트 번호, 민감한 정보 같은 설정을 코드에 하드코딩하는 걸 막아 줘요.

  • 설정을 코드 밖으로 분리할 수 있어요.

  • 코드가 더 유연해지고 재사용하기 좋아집니다.

아래 두 섹션에서 모든 옵션을 나열해요. 컴포넌트 옵션을 먼저 보여 주고, 그다음에 엔드포인트 옵션을 보여 줍니다.

컴포넌트 옵션 (Component Options)

Debezium MySQL Connector 컴포넌트가 지원하는 옵션은 아래와 같아요.

Name Description Default Type
additionalProperties (common) debezium 컴포넌트 설정에 직접 담을 수 없는 추가 프로퍼티를 넣기 위한 옵션이에요(예: Debezium 엔진에 필요한 Kafka Connect 프로퍼티 설정, 예컨대 KafkaOffsetBackingStore 설정). 프로퍼티 앞에는 additionalProperties. 접두사를 붙여야 해요. 예: additionalProperties.transactional.id=12345&additionalProperties.schema.registry.url=http://localhost:8811/avro. additionalProperties. 접두사를 쓰는 다중 값(multi-value) 옵션이에요. Map
bridgeErrorHandler (consumer) 컨슈머를 Camel 라우팅 에러 핸들러(Error Handler)에 브리지할 수 있게 해 줘요. 즉, Camel 컨슈머가 들어오는 메시지를 집어 올리는 동안 등에서 예외가 발생하면(가능한 경우), 그 예외를 메시지로 처리해서 라우팅 에러 핸들러가 다루게 돼요. 중요: 이것은 서드파티 컴포넌트가 예외가 던져졌을 때 Camel에 알릴 수 있도록 허용하는 경우에만 가능해요. 일부 컴포넌트는 내부에서만 처리하기 때문에 bridgeErrorHandler를 쓸 수 없어요. 다른 상황에서는 Camel 컴포넌트를 개선해서 서드파티 컴포넌트에 연결(hook)해 이를 가능하게 만들 수도 있어요. 기본으로 컨슈머는 org.apache.camel.spi.ExceptionHandler로 예외를 처리하는데, WARN 또는 ERROR 레벨로 로그를 남기고 무시해요. false boolean
configuration (consumer) 미리 구성된 설정을 지정할 수 있게 해 줘요. MySqlConnectorEmbeddedDebeziumConfiguration
internalKeyConverter (consumer) 오프셋의 키 데이터를 직렬화·역직렬화하는 데 쓸 Converter 클래스예요. 기본값은 JSON converter예요. org.apache.kafka.connect.json.JsonConverter String
internalValueConverter (consumer) 오프셋의 값 데이터를 직렬화·역직렬화하는 데 쓸 Converter 클래스예요. 기본값은 JSON converter예요. org.apache.kafka.connect.json.JsonConverter String
offsetCommitPolicy (consumer) 커밋 정책(commit policy)의 Java 클래스 이름이에요. 처리한 이벤트 수와 마지막 커밋 이후 경과 시간을 기준으로 오프셋 커밋을 언제 트리거할지 정의해요. 이 클래스는 'OffsetCommitPolicy' 인터페이스를 구현해야 하고, 기본값은 시간 간격 기반의 주기적 커밋 정책이에요. String
offsetCommitTimeoutMs (consumer) 레코드를 플러시하고 파티션 오프셋 데이터를 오프셋 스토리지에 커밋하기까지 기다릴 최대 밀리초예요. 이 시간을 넘으면 프로세스를 취소하고, 해당 오프셋 데이터 커밋은 나중에 다시 시도하도록 복원해요. 기본값은 5초입니다. 5000 long
offsetFlushIntervalMs (consumer) 오프셋 커밋을 시도하는 간격이에요. 기본값은 1분입니다. 60000 long
offsetStorage (consumer) 커넥터 오프셋을 영속화하는 일을 맡는 Java 클래스 이름이에요. org.apache.kafka.connect.storage.FileOffsetBackingStore String
offsetStorageFileName (consumer) 오프셋을 저장할 파일 경로예요. offset.storage가 FileOffsetBackingStore로 설정된 경우 필요해요. String
offsetStoragePartitions (consumer) 오프셋 스토리지 토픽을 만들 때 쓰는 파티션 수예요. offset.storage가 'KafkaOffsetBackingStore'로 설정된 경우 필요합니다. int
offsetStorageReplicationFactor (consumer) 오프셋 스토리지 토픽을 만들 때 쓰는 복제 계수(replication factor)예요. offset.storage가 KafkaOffsetBackingStore로 설정된 경우 필요합니다. int
offsetStorageTopic (consumer) 오프셋을 저장할 Kafka 토픽 이름이에요. offset.storage가 KafkaOffsetBackingStore로 설정된 경우 필요해요. String
autowiredEnabled (advanced) 오토와이어링(autowiring)을 켤지 여부예요. 레지스트리에서 타입이 맞는 인스턴스가 딱 하나 있는지 찾아서 컴포넌트에 자동으로 주입하는(옵션이 autowired로 표시돼 있어야 함) 기능에 쓰여요. JDBC 데이터 소스, JMS 커넥션 팩토리, AWS 클라이언트 등을 자동 구성할 때 활용할 수 있습니다. true boolean
bigintUnsignedHandlingMode (mysql) BIGINT UNSIGNED 컬럼을 변경 이벤트에 어떻게 표현할지 정해요. 'precise'는 java.math.BigDecimal로 값을 표현하고, 변경 이벤트에는 이진 표현과 Kafka Connect의 'org.apache.kafka.connect.data.Decimal' 타입으로 인코딩돼요. 'long'(기본값)은 Java의 'long'으로 값을 표현하는데, 정밀도는 떨어질 수 있지만 consumer에서 훨씬 다루기 쉬워요. long String
binlogBufferSize (mysql) binlog reader가 진행 중인 트랜잭션이 커밋될지 롤백될지 판단하려고 쓰는 룩어헤드(look-ahead) 버퍼 크기예요. 0으로 두면 룩어헤드 버퍼링을 끕니다. 기본값은 0(버퍼링 꺼짐)이에요. 0 int
binlogNetReadTimeout (mysql) binlog 연결에서 읽기가 끝나기를 서버가 타임아웃 전까지 기다리는 초예요. 0이면 MySQL 서버 기본값을 씁니다. 스트리밍 중 EOFException을 막으려면 지연이 큰 네트워크 환경에서 이 값을 늘려야 할 수 있어요. 0 long
binlogNetWriteTimeout (mysql) binlog 연결에 쓰기가 끝나기를 서버가 타임아웃 전까지 기다리는 초예요. 0이면 MySQL 서버 기본값을 씁니다. 대용량 데이터로 스트리밍 중 EOFException이 나면 이 값을 늘려야 할 수 있어요. 0 long
columnExcludeList (mysql) 변경 이벤트에서 제외할 컬럼과 맞아떨어지는 정규식이에요. String
columnIncludeList (mysql) 변경 이벤트에 포함할 컬럼과 맞아떨어지는 정규식이에요. String
columnPropagateSourceType (mysql) 컬럼의 완전한 이름(fully-qualified name)과 맞는 정규식을 콤마로 나열한 값이에요. 이에 맞는 컬럼은 원래 타입과 원래 길이를 파라미터로 내보내는 변경 레코드의 필드 스키마에 추가돼요. String
connectionValidationTimeoutMs (mysql) 연결 검증이 끝나기를 기다리는 최대 시간(밀리초)이에요. 기본값은 60초입니다. 1m long
connectKeepAlive (mysql) 연결이 살아 있도록 유지하려고 별도 스레드를 쓸지 여부예요. true boolean
connectKeepAliveIntervalMs (mysql) keep alive 스레드를 쓸 때 연결을 확인하는 간격(밀리초)이에요. 기본값은 1분(60,000 ms)입니다. 1m long
connectTimeoutMs (mysql) 데이터베이스에 연결을 시도한 뒤 타임아웃되기까지 기다리는 최대 시간(밀리초)이에요. 기본값은 30초(30,000 ms)입니다. 30s int
converters (mysql) 기본 컨버터 대신 쓸 커스텀 컨버터 목록(선택)이에요. 컨버터는 '.type' 설정 옵션으로 정의하고 '.' 옵션으로 구성합니다. String
customMetricTags (mysql) 커스텀 메트릭 태그는 키-값 쌍을 받아서 MBean 객체 이름을 커스터마이즈해요. 이 이름은 일반 이름 끝에 붙고, 각 키는 MBean 객체 이름의 태그가 되며 그 값은 해당 태그의 값이 됩니다. 예: k1=v1,k2=v2. String
customSanitizePattern (mysql) 값을 마스킹할 설정 키를 골라내는 정규식이에요. 설정하면 Debezium의 기본 비밀번호 마스킹 패턴 대신 이 커스텀 패턴을 씁니다. .*secret$ .*password$
databaseExcludeList (mysql) 모니터링에서 제외할 데이터베이스 이름과 맞는 정규식을 콤마로 나열한 값이에요. String
databaseHostname (mysql) 데이터베이스 서버의 DNS로 풀리는 호스트 이름이나 IP 주소예요. String
databaseIncludeList (mysql) 변경을 캡처할 데이터베이스들이에요. String
databaseInitialStatements (mysql) 데이터베이스에 JDBC 연결(binlog 읽기 연결 제외)이 맺어질 때 실행할 SQL 문을 세미콜론으로 나열한 값이에요. 커넥터가 임의로 JDBC 연결을 맺을 수 있다는 점에 유의하세요. 그래서 보통은 DML 문 실행이 아니라 세션 파라미터 설정 용도로만 쓰는 게 좋습니다. 세미콜론을 구분자가 아닌 문자로 쓰려면 (';;')처럼 두 개를 겹쳐 쓰세요. String
databaseJdbcDriver (mysql) MySQL 데이터베이스 서버에 연결할 때 쓰는 JDBC 드라이버 클래스 이름이에요. com.mysql.cj.jdbc.Driver String
databasePassword (mysql) 필수 데이터베이스에 연결할 때 쓸 데이터베이스 사용자의 비밀번호예요. String
databasePort (mysql) 데이터베이스 서버의 포트예요. 3306 int
databaseProtocol (mysql) 드라이버와 함께 쓸 JDBC 프로토콜이에요. jdbc:mysql String
databaseQueryTimeoutMs (mysql) 쿼리 실행을 기다리는 시간(밀리초)이에요. 기본값은 600초(600,000 ms)이고, 0이면 제한이 없어요. 10m int
databaseServerId (mysql) 이 데이터베이스 클라이언트의 숫자 ID예요. 클러스터에서 현재 도는 모든 데이터베이스 프로세스 사이에서 유일해야 합니다. 커넥터는 이 고유 ID를 가진 또 하나의 서버로서 데이터베이스 클러스터에 합류해서 binlog를 읽어요. long
databaseServerIdOffset (mysql) 병렬 스냅샷(parallel snapshotting)을 설정한 경우에만 관련 있어요. 병렬 스냅샷 중에는 데이터베이스 클라이언트에 여러(4개) 연결이 열리고, 각 연결은 고유한 연결 ID가 필요해요. 이 오프셋은 기준이 되는 클러스터 ID로부터 그 ID들을 만들어 낼 때 씁니다. 10000 long
databaseSslKeystore (mysql) 키 스토어 파일의 위치예요. 선택 사항이고, 클라이언트와 데이터베이스 사이의 양방향 인증(two-way authentication)에 쓸 수 있어요. String
databaseSslKeystorePassword (mysql) 키 스토어 파일의 비밀번호예요. 선택 사항이고, 'database.ssl.keystore'를 설정한 경우에만 필요합니다. String
databaseSslMode (mysql) 데이터베이스에 암호화 연결을 쓸지 정해요. 옵션: 'disabled'는 암호화하지 않은 연결을 써요. 'preferred'(기본값)는 서버가 보안 연결을 지원하면 보안(암호화) 연결을 맺고, 아니면 암호화하지 않은 연결로 물러나요. 'required'는 보안(암호화) 연결을 쓰되, 맺을 수 없으면 실패해요. 'verify_ca'는 'required'에 더해 설정된 인증 기관(CA) 인증서로 서버 TLS 인증서를 검증하고, 맞는 유효한 CA 인증서가 없으면 실패해요. 'verify_identity'는 'verify_ca'에 더해 서버 인증서가 접속하려는 호스트와 맞는지도 검증합니다. preferred String
databaseSslTruststore (mysql) 서버 인증서 검증에 쓸 트러스트 스토어 파일의 위치예요. String
databaseSslTruststorePassword (mysql) 트러스트 스토어 파일의 비밀번호예요. 트러스트스토어 무결성 검사와 잠금 해제에 씁니다. String
databaseUser (mysql) 데이터베이스에 연결할 때 쓸 데이터베이스 사용자 이름이에요. String
datatypePropagateSourceType (mysql) 데이터베이스별 데이터 타입 이름과 맞는 정규식을 콤마로 나열한 값이에요. 이에 맞는 데이터 타입은 원래 타입과 원래 길이를 파라미터로 내보내는 변경 레코드의 필드 스키마에 추가돼요. String
ddlParserType (mysql) MySQL DDL 문을 파싱할 때 어떤 ANTLR 문법을 쓸지 정해요. 'default'는 Oracle MySQL 문법을 쓰는데, 꾸준히 관리되고 MySQL 8.0 기능을 지원해요. 'legacy'는 기존 배포 환경과의 하위 호환을 위해 Positive Technologies 문법을 씁니다. default String
decimalHandlingMode (mysql) DECIMAL과 NUMERIC 컬럼을 변경 이벤트에 어떻게 표현할지 정해요. 'precise'(기본값)는 java.math.BigDecimal로 값을 표현하고, 변경 이벤트에는 이진 표현과 Kafka Connect의 'org.apache.kafka.connect.data.Decimal' 타입으로 인코딩돼요. 'string'은 문자열로 값을 표현해요. 'double'은 Java의 'double'로 값을 표현하는데, 정밀도는 떨어질 수 있지만 consumer에서 훨씬 다루기 쉬워요. precise String
enableTimeAdjuster (mysql) 데이터베이스는 연도 값을 2자리 또는 4자리로 넣게 허용해요. 2자리면 값이 1970~2069 범위로 자동 매핑돼요. false는 암시적 변환을 데이터베이스에 맡기고, true(기본값)는 Debezium이 변환을 수행해요. true boolean
errorsMaxRetries (mysql) 실패하기 전까지 연결 오류를 재시도할 최대 횟수예요(-1 = 제한 없음, 0 = 비활성화, 그 외 = 재시도 횟수). -1 int
eventDeserializationFailureHandlingMode (mysql) binlog 이벤트 역직렬화 중 장애(즉, 손상된 이벤트를 만났을 때)를 어떻게 처리할지 정해요. 'fail'(기본값)은 문제 이벤트와 그 binlog 위치를 알리는 예외를 던져 커넥터를 멈춰요. 'warn'은 문제 이벤트와 binlog 위치를 로그로 남기고 이벤트를 건너뛰어요. 'ignore'는 문제 이벤트를 조용히 건너뜁니다. fail String
eventProcessingFailureHandlingMode (mysql) 이벤트 처리 중 장애(즉, 손상된 이벤트를 만났을 때)를 어떻게 처리할지 정해요. 'fail'(기본값)은 문제 이벤트와 그 위치를 알리는 예외를 던져 커넥터를 멈춰요. 'warn'은 문제 이벤트와 위치를 로그로 남기고 이벤트를 건너뛰어요. 'ignore'는 문제 이벤트를 조용히 건너뜁니다. fail String
executorShutdownTimeoutMs (mysql) 태스크 익스큐터(task executor)가 꺼지기를 기다리는 최대 시간(밀리초)이에요. 4s long
extendedHeadersEnabled (mysql) 다운스트림 처리 시스템에서 CDC 이벤트의 출처를 추적·식별하는 데 꼭 필요한 메타데이터를 담은 Debezium 컨텍스트 헤더를 켜거나 꺼요. true boolean
gtidIgnoreOnRecovery (mysql) 복구 중에 GTID를 무시하고 대신 binlog 파일과 위치부터 다시 시작할지 여부예요. 서버의 GTID 모드는 그대로 켜져 있고, 복구 후에는 GTID 추적이 정상으로 돌아옵니다. false boolean
gtidSourceExcludes (mysql) MySQL 서버 binlog에서 시작 위치를 정할 때 GTID 범위를 제외하는 데 쓰는 소스 UUID들이에요. String
gtidSourceFilterDmlEvents (mysql) true로 두면, gtid.source.includes나 gtid.source.excludes(설정한 경우)가 정의한 GTID와 맞는 트랜잭션만 골라서 DML 이벤트를 만들어요. true boolean
gtidSourceIncludes (mysql) MySQL 서버 binlog에서 시작 위치를 정할 때 GTID 범위를 포함하는 데 쓰는 소스 UUID들이에요. String
guardrailCollectionsLimitAction (mysql) guardrail 컬렉션 한도를 넘었을 때 취할 동작을 정해요. 'warn'(기본값)은 경고 메시지를 로그로 남기고 처리를 계속하고, 'fail'은 에러로 커넥터를 멈춥니다. warn String
guardrailCollectionsMax (mysql) 커넥터가 캡처할 수 있는 컬렉션·테이블의 최대 개수예요. 한도를 넘으면 'guardrail.collections.limit.action'이 정한 동작을 합니다. 이 guardrail을 끄려면 0으로 두세요. 0 int
heartbeatActionQuery (mysql) 하트비트마다 실행할 쿼리예요. String
heartbeatIntervalMs (mysql) 커넥터가 하트비트 토픽에 주기적으로 하트비트 메시지를 보내는 간격(밀리초)이에요. 0으로 두면 하트비트 메시지를 끕니다. 기본은 꺼져 있어요. 0ms int
heartbeatTopicsPrefix (mysql) 하트비트 토픽 이름에 쓰는 접두사예요. 기본값은 __debezium-heartbeat입니다. __debezium-heartbeat String
includeQuery (mysql) 변경 이벤트를 만든 원본 SQL 쿼리를 이벤트에 포함할지 여부예요. 참고: 이 옵션을 쓰려면 데이터베이스가 서버 옵션 binlog_rows_query_log_events(MySQL) 또는 binlog_annotate_row_events(MariaDB)를 ON으로 설정된 상태여야 해요. 스냅샷에서 만든 이벤트에는 쿼리가 담기지 않아요. 경고: 이 옵션을 켜면 변경 이벤트에 원본 SQL 문이 담기면서, 의도적으로 제외하거나 마스킹한 테이블·필드가 노출될 수 있어요. 그래서 기본값은 'false'입니다. false boolean
includeSchemaChanges (mysql) 데이터베이스 스키마 변경을 데이터베이스 서버 ID와 같은 이름의 Kafka 토픽에 발행할지 여부예요. 각 스키마 변경은 데이터베이스 이름을 담은 키로 기록되고, 값에는 새 스키마의 논리적 설명과 (있다면) DDL 문이 들어가요. 기본값은 'true'입니다. 이는 커넥터가 내부적으로 데이터베이스 스키마 히스토리를 기록하는 방식과는 별개예요. true boolean
includeSchemaComments (mysql) 테이블·컬럼 주석을 메타데이터 객체로 파싱할지 여부예요. 참고: 이 옵션을 켜면 메모리 사용량에 영향이 있어요. Debezium 커넥터의 메모리 소비는 ColumnImpl 객체의 개수와 크기가 좌우하는데, 객체마다 String을 하나씩 붙이면 꽤 무거워질 수 있습니다. 기본값은 'false'예요. false boolean
inconsistentSchemaHandlingMode (mysql) 내부 스키마 표현에 없는 테이블(즉, 내부 표현이 데이터베이스와 일치하지 않음)에 속한 binlog 이벤트를 어떻게 처리할지 정해요. 'fail'(기본값)은 문제 이벤트와 그 binlog 위치를 알리는 예외를 던져 커넥터를 멈춰요. 'warn'은 문제 이벤트와 binlog 위치를 로그로 남기고 이벤트를 건너뛰어요. 'skip'은 문제 이벤트를 조용히 건너뜁니다. fail String
incrementalSnapshotAllowSchemaChanges (mysql) 증분 스냅샷 중 스키마 변경을 감지하고 현재 청크를 다시 골라서 잠금(locking) DDL을 피하게 해 줘요. 기본 키 변경은 지원하지 않고, 증분 스냅샷 중에 수행하면 잘못된 결과를 낼 수 있다는 점에 유의하세요. 또한 스키마 변경이 컬럼 기본값에만 영향을 준다면, binlog 스트림에서 DDL을 처리하기 전까지는 변경을 감지하지 못해요. 이는 스냅샷 이벤트의 값에는 영향을 주지 않지만, 스냅샷 이벤트의 스키마에는 오래된 기본값이 남을 수 있습니다. false boolean
incrementalSnapshotChunkSize (mysql) 증분 스냅샷에서 청크(문서/행 개수)의 최대 크기예요. 1024 int
incrementalSnapshotWatermarkingStrategy (mysql) 증분 스냅샷 동안 워터마킹(watermarking)에 쓸 전략을 정해요. 'insert_insert'는 열기(open)와 닫기(close) 시그널을 모두 시그널 데이터 컬렉션에 기록해요(기본값). 'insert_delete'는 열기 시그널만 시그널 데이터 컬렉션에 기록하고, 닫기가 상응하는 열기 시그널을 지웁니다. INSERT_INSERT String
legacySnapshotMaxThreads (mysql) 켜면 구 방식(table-per-thread) 병렬 스냅샷 알고리즘을 써요. false(기본값)로 두면 테이블을 청크로 쪼개 모든 스냅샷 스레드에 나눠 처리해서, 스냅샷 동시성을 더 높일 수 있어요. false boolean
maxBatchSize (mysql) 소스 레코드 배치 하나의 최대 크기예요. 기본값은 2048입니다. 2048 int
maxQueueSize (mysql) 데이터베이스 로그에서 읽었지만 아직 기록·전달되지 않은 변경 이벤트 큐의 최대 크기예요. 기본값은 8192이고, 항상 최대 배치 크기보다 커야 합니다. 8192 int
maxQueueSizeInBytes (mysql) 데이터베이스 로그에서 읽었지만 아직 기록·전달되지 않은 변경 이벤트 큐의 최대 크기(바이트)예요. 기본값은 0이고, 이는 기능이 꺼져 있다는 뜻이에요. 0 long
memoryManagementSchemasClass (mysql) 스키마 메타데이터 저장 구현체의 완전한 클래스 이름(fully-qualified)이에요. 이 클래스는 io.debezium.relational.TableMappingStorage를 구현해야 합니다. 기본값은 인메모리 저장용 io.debezium.relational.ConcurrentMapTableMappingStorage예요. io.debezium.relational.ConcurrentMapTableMappingStorage String
memoryManagementTablesClass (mysql) 테이블 메타데이터 저장 구현체의 완전한 클래스 이름(fully-qualified)이에요. 이 클래스는 io.debezium.relational.TableMappingStorage를 구현해야 합니다. 기본값은 인메모리 저장용 io.debezium.relational.ConcurrentMapTableMappingStorage예요. io.debezium.relational.ConcurrentMapTableMappingStorage String
messageKeyColumns (mysql) 메시지 키로 쓸 완전한 이름(fully-qualified)의 테이블·컬럼과 맞는 표현식을 세미콜론으로 나열한 값이에요. 각 표현식은 ':' 패턴과 일치해야 하고, 테이블 이름은 커넥터에 따라 (DB_NAME.TABLE_NAME) 또는 (SCHEMA_NAME.TABLE_NAME)으로 정의할 수 있어요. 키 컬럼은 커스텀 키가 되는 컬럼들을 콤마로 나열한 값입니다. 명시적으로 키 설정이 없는 테이블은 그 테이블의 기본 키 컬럼이 메시지 키로 쓰여요. 예: dbserver1.inventory.orderlines:orderId,orderLineId;dbserver1.inventory.orders:id. String
minRowCountToStreamResults (mysql) 스냅샷 동안 결과를 전부 메모리로 당겨 오는 대신 스트리밍하려면, 테이블이 가져야 하는 최소 행 수예요. 기본값은 1,000입니다. 0으로 두면 모든 결과를 스트리밍해서 테이블 크기 확인을 아예 생략해요. 1000 int
notificationEnabledChannels (mysql) 활성화할 알림(notifications) 채널 이름 목록이에요. String
notificationSinkTopicName (mysql) 알림을 보낼 토픽 이름이에요. 활성 채널 목록에 'sink'가 있으면 필수입니다. String
openlineageIntegrationConfigFilePath (mysql) OpenLineage 설정 파일의 경로예요. https://openlineage.io/docs/client/java/configuration을 참고하세요. ./openlineage.yml String
openlineageIntegrationDatasetKafkaBootstrapServers (mysql) 입력/출력 네임스페이스로 쓰는 Kafka 부트스트랩 서버 주소예요. String
openlineageIntegrationEnabled (mysql) Debezium이 OpenLineage API로 데이터 계보(data lineage) 메타데이터를 내보내도록 켜요. false boolean
openlineageIntegrationJobDescription (mysql) Debezium이 내보내는 잡(job) 설명이에요. Debezium change data capture job String
openlineageIntegrationJobNamespace (mysql) Debezium이 내보내는 잡 네임스페이스예요. String
openlineageIntegrationJobOwners (mysql) Debezium이 내보내는 잡 소유자예요. 키-값 쌍을 콤마로 나열합니다. 예: k1=v1,k2=v2. String
openlineageIntegrationJobTags (mysql) Debezium이 내보내는 잡 태그예요. 키-값 쌍을 콤마로 나열합니다. 예: k1=v1,k2=v2. String
pollIntervalMs (mysql) 이벤트가 오지 않은 뒤 새 변경 이벤트를 기다리는 시간(밀리초)이에요. 기본값은 500 ms입니다. 500ms long
postProcessors (mysql) 포스트 프로세서 목록(선택)이에요. 프로세서는 '.type' 설정 옵션으로 정의하고 '' 옵션으로 구성합니다. String
provideTransactionMetadata (mysql) 이벤트 카운트와 함께 트랜잭션 메타데이터 추출을 켜요. false boolean
queryFetchSize (mysql) 스트리밍 동안 메모리로 가져올 레코드의 최대 개수예요. '0'이면 기본 JDBC fetch size를 씁니다. 0 int
retriableRestartConnectorWaitMs (mysql) 재시도 가능한 예외가 발생한 뒤 커넥터를 다시 시작하기 전에 기다리는 시간이에요. 기본값은 10000ms입니다. 10s long
schemaHistoryInternal (mysql) 데이터베이스 스키마 변경을 저장·복구하는 데 쓸 SchemaHistory 클래스 이름이에요. 히스토리 관련 설정 프로퍼티는 'schema.history.internal.' 문자열을 접두사로 씁니다. io.debezium.storage.kafka.history.KafkaSchemaHistory String
schemaHistoryInternalFileFilename (mysql) 데이터베이스 스키마 히스토리를 기록할 파일의 경로예요. String
schemaHistoryInternalSkipUnparseableDdl (mysql) binlog에서 파싱할 수 없는 DDL 문을 만났을 때 Debezium이 취할 동작을 정해요. 기본으로 커넥터는 동작을 멈추지만, 설정을 바꾸면 파싱할 수 없는 문을 무시할 수 있어요. 건너뛰기를 켜면 Debezium이 메타데이터 변경을 놓칠 수 있습니다. false boolean
schemaHistoryInternalStoreOnlyCapturedDatabasesDdl (mysql) Debezium이 데이터베이스 스키마 히스토리에 어떤 DDL을 저장할지 정해요. 기본(false)은 들어오는 모든 DDL 문을 저장해요. true로 두면 캡처한 스키마/데이터베이스의 테이블을 건드리는 DDL만 저장합니다. false boolean
schemaHistoryInternalStoreOnlyCapturedTablesDdl (mysql) Debezium이 데이터베이스 스키마 히스토리에 어떤 DDL을 저장할지 정해요. 기본(false)은 들어오는 모든 DDL 문을 저장해요. true로 두면 캡처한 테이블을 건드리는 DDL만 저장합니다. false boolean
schemaNameAdjustmentMode (mysql) 커넥터가 쓰는 메시지 컨버터와 호환되도록 스키마 이름을 어떻게 조정할지 정해요. 'avro'는 Avro 타입 이름에 쓸 수 없는 문자를 밑줄(underscore)로 바꿔요. 'avro_unicode'는 Avro 타입 이름에 쓸 수 없는 문자나 밑줄을 _uxxxx 같은 해당 유니코드로 바꿉니다. 참고: _는 Java의 백슬래시 같은 이스케이프 시퀀스예요. 'none'은 아무 조정도 하지 않아요(기본값). none String
signalDataCollection (mysql) Debezium에 시그널/명령을 보낼 때 쓰는 데이터 컬렉션 이름이에요. 멀티 파티션 모드 커넥터는 시그널 데이터 컬렉션을 콤마로 구분한 목록으로 여러 개 지정할 수 있어요. 설정하지 않으면 시그널링이 꺼집니다. String
signalEnabledChannels (mysql) 활성화할 시그널(signal) 채널 이름 목록이에요. source 채널이 기본으로 켜져 있습니다. source String
signalPollIntervalMs (mysql) 등록된 채널에서 새 시그널을 찾는 간격(밀리초)이에요. 기본값은 5초입니다. 5s long
skippedOperations (mysql) 스트리밍 중 건너뛸 연산을 콤마로 나열한 값이에요. 'c'는 insert/create, 'u'는 update, 'd'는 delete, 't'는 truncate, 'none'은 건너뛰지 않음을 뜻해요. 기본으로는 truncate 연산만 건너뜁니다. t String
snapshotDelayMs (mysql) 스냅샷이 시작되기 전에 기다리는 지연 시간(밀리초)이에요. 기본값은 0 ms입니다. 0ms long
snapshotFetchSize (mysql) 스냅샷을 수행하는 동안 메모리로 가져올 레코드의 최대 개수예요. int
snapshotIncludeCollectionList (mysql) 커넥터를 만들거나 다시 시작할 때 스냅샷을 찍을 테이블/컬렉션 목록을 지정하려면 이 설정을 반드시 써야 해요. String
snapshotLockingMode (mysql) 스냅샷을 수행하는 동안 커넥터가 전역 읽기 잠금(global read lock)을 얼마나 오래 쥐고 있을지 정해요. 기본값인 'minimal'은 데이터베이스 스키마 등 다른 메타데이터를 읽는 스냅샷 초기 구간 동안만 전역 읽기 잠금을 쥐고(그래서 업데이트를 막고) 있어요. 스냅샷의 나머지 작업은 각 테이블에서 모든 행을 읽는 건데, 잠금을 놓은 뒤에도 다른 연산이 데이터베이스를 갱신하는 동안 스냅샷 프로세스의 REPEATABLE READ 트랜잭션으로 처리할 수 있어요. 하지만 어떤 경우에는 스냅샷 내내 모든 쓰기를 막는 게 좋을 수도 있어요. 그럴 때는 이 프로퍼티를 'extended'로 두세요. 'none'으로 두면 스냅샷 과정에서 커넥터가 어떤 테이블 잠금도 걸지 않아요. 이 모드는 snapshot.mode가 'schema_only' 또는 'schema_only_recovery'일 때만 조합할 수 있고, 스냅샷을 찍는 동안 스키마 변경이 일어나지 않을 때만 안전합니다. minimal String
snapshotLockTimeoutMs (mysql) 스냅샷 시작 시 테이블 잠금을 기다리는 최대 밀리초예요. 이 시간 안에 잠금을 얻지 못하면 스냅샷이 중단돼요. 기본값은 10초입니다. 10s long
snapshotMaxThreads (mysql) 스냅샷을 수행하는 데 쓰는 스레드의 최대 개수예요. 기본값은 1입니다. 1 int
snapshotMaxThreadsMultiplier (mysql) 테이블별 스냅샷 청크 개수를 조절하는 배율이에요. 기본 동작은 'row_count/snapshot.max.threads'로 청크당 행 수를 계산해요. 큰 테이블에서는 이게 최선이 아닐 수 있는데, 배율을 쓰면 'row_count/(snapshot.max.threads snapshot.max.threads.multiplier)' 공식으로 청크 수를 늘립니다. 1 int
snapshotMode (mysql) 커넥터 시작 시 스냅샷을 실행하는 기준이에요. 다음 스냅샷 옵션 중 하나를 고르세요. 'when_needed': 시작할 때 스냅샷이 필요하면 실행해요. 'schema_only': 논리적 서버 이름에 대한 오프셋이 없으면 테이블 구조(스키마)만 캡처하는 스냅샷을 실행하고, 테이블 데이터는 담지 않아요. 스냅샷이 끝나면 binlog에서 변경을 스트리밍하기 시작합니다. 'schema_only_recovery': 데이터베이스 스키마 히스토리만 캡처하는 스냅샷을 수행하고, 다시 스트리밍으로 돌아가요. 손상되거나 잃어버린 데이터베이스 스키마 히스토리 토픽을 복구할 때 쓰세요. 커넥터가 멈춘 뒤에 데이터베이스 스키마가 바뀌었다면 쓰지 마세요. 'initial'(기본값): 논리적 서버 이름에 대한 오프셋이 없으면 설정한 테이블의 현재 전체 상태를 캡처하는 스냅샷을 실행해요. 스냅샷이 끝나면 binlog에서 변경을 스트리밍하기 시작합니다. 'initial_only': 'initial'처럼 스냅샷을 수행하지만, 스냅샷이 끝나면 멈추고 binlog에서 변경을 스트리밍하지 않아요. initial String
snapshotModeConfigurationBasedSnapshotData (mysql) 'snapshot.mode'가 configuration_based로 설정됐을 때, 데이터를 스냅샷에 담을지 여부를 정하는 설정이에요. false boolean
snapshotModeConfigurationBasedSnapshotOnDataError (mysql) 'snapshot.mode'가 configuration_based로 설정됐을 때, 에러가 났을 경우 데이터를 스냅샷에 담을지 여부를 정하는 설정이에요. false boolean
snapshotModeConfigurationBasedSnapshotOnSchemaError (mysql) 'snapshot.mode'가 configuration_based로 설정됐을 때, 에러가 났을 경우 스키마를 스냅샷에 담을지 여부를 정하는 설정이에요. false boolean
snapshotModeConfigurationBasedSnapshotSchema (mysql) 'snapshot.mode'가 configuration_based로 설정됐을 때, 스키마를 스냅샷에 담을지 여부를 정하는 설정이에요. false boolean
snapshotModeConfigurationBasedStartStream (mysql) 'snapshot.mode'가 configuration_based로 설정됐을 때, 스냅샷 뒤에 스트리밍을 시작할지 여부를 정하는 설정이에요. false boolean
snapshotModeCustomName (mysql) 'snapshot.mode'가 custom으로 설정됐을 때는 반드시 설정해야 해요. 'name()' 메서드가 제공하는 커스텀 구현의 이름을 지정합니다. 구현체는 'Snapshotter' 인터페이스를 구현해야 하고, 앱이 부팅될 때마다 스냅샷을 찍을지 판단하려고 호출돼요. String
snapshotQueryMode (mysql) 스냅샷 동안 쓸 쿼리를 제어해요. select_all String
snapshotQueryModeCustomName (mysql) 'snapshot.query.mode'가 custom으로 설정됐을 때는 반드시 설정해야 해요. 'name()' 메서드가 제공하는 커스텀 구현의 이름을 지정합니다. 구현체는 'SnapshotterQuery' 인터페이스를 구현해야 하고, 스냅샷 동안 쿼리를 어떻게 만들지 결정하려고 호출돼요. String
snapshotSelectStatementOverrides (mysql) 이 프로퍼티는 완전한 이름(fully-qualified)의 테이블 (DB_NAME.TABLE_NAME) 또는 (SCHEMA_NAME.TABLE_NAME)을 콤마로 나열한 값이에요(커넥터에 따라 다름). 테이블별 select 문은 이어지는 설정 프로퍼티에서 지정해요. 각 테이블마다 하나씩, 'snapshot.select.statement.overrides.DB_NAME.TABLE_NAME' 또는 'snapshot.select.statement.overrides.SCHEMA_NAME.TABLE_NAME' 아이디로 식별합니다. 그 프로퍼티 값은 스냅샷 중 해당 테이블에서 데이터를 읽을 때 쓸 select 문이에요. 대용량 append-only 테이블이라면, 이전 스냅샷이 중단됐을 때 스냅샷을 (재개) 시작할 지점을 정하는 용도로 쓸 수 있어요. String
snapshotTablesOrderByRowCount (mysql) 초기 스냅샷에서 테이블을 처리하는 순서를 제어해요. descending은 행 수 내림차순으로, ascending은 행 수 오름차순으로 테이블을 정렬해요. disabled(기본값)는 행 수 정렬을 끕니다. disabled String
sourceinfoStructMaker (mysql) SourceInfo 스키마와 struct를 돌려주는 SourceInfoStructMaker 클래스 이름이에요. io.debezium.connector.mysql.MySqlSourceInfoStructMaker String
statisticsMetricsEnabled (mysql) 레코드 처리 지연 시간 같은 여러 통계와, 퀀타일(quantiles) 같은 파생 데이터 수집을 켜요. 기본으로 통계 수집은 켜져 있습니다. true boolean
streamingDelayMs (mysql) 스냅샷이 끝나고 스트리밍이 시작되기 전까지의 지연 시간(밀리초)이에요. 기본값은 0 ms입니다. 0ms long
tableExcludeList (mysql) 모니터링에서 제외할 테이블의 완전한 이름(fully-qualified)과 맞는 정규식을 콤마로 나열한 값이에요. String
tableIgnoreBuiltin (mysql) 기본 제공(built-in) 테이블을 무시할지 정하는 플래그예요. true boolean
tableIncludeList (mysql) 변경을 캡처할 테이블들이에요. String
timePrecisionMode (mysql) 시간, 날짜, 타임스탬프는 여러 정밀도로 표현할 수 있어요. 'adaptive_time_microseconds': 날짜·타임스탬프 값의 정밀도는 데이터베이스 컬럼의 정밀도를 따르지만, time 필드는 항상 마이크로초 정밀도를 씁니다. 'connect': 데이터베이스 컬럼 정밀도와 무관하게 항상 밀리초 정밀도로, Kafka Connect의 Time·Date·Timestamp 내장 표현으로 시간·날짜·타임스탬프 값을 표현해요. adaptive_time_microseconds String
tombstonesOnDelete (mysql) 삭제 연산을 delete 이벤트와 그 뒤의 tombstone 이벤트로 나타낼지(true), delete 이벤트만으로 나타낼지(false) 정해요. tombstone 이벤트를 내보내면(기본 동작) 소스 레코드가 지워진 뒤 Kafka가 해당 키와 관련된 이벤트를 전부 지울 수 있어요. false boolean
topicNamingStrategy (mysql) 데이터 변경, 스키마 변경, 트랜잭션, 하트비트 이벤트 등의 토픽 이름을 정하는 데 쓸 TopicNamingStrategy 클래스 이름이에요. io.debezium.schema.SchemaTopicNamingStrategy String
topicPrefix (mysql) 필수 변경을 캡처하는 해당 데이터베이스 서버/클러스터를 식별하고 네임스페이스를 제공하는 토픽 접두사예요. 이 커넥터가 내보내는 이벤트를 받는 모든 Kafka 토픽 이름의 접두사로 쓰이므로, 다른 모든 커넥터 사이에서 유일해야 합니다. 영숫자와 하이픈, 점, 밑줄만 받아들이세요. String
transactionMetadataFactory (mysql) 트랜잭션 컨텍스트와 트랜잭션 struct/스키마를 만드는 클래스예요. io.debezium.pipeline.txmetadata.DefaultTransactionMetadataFactory String
useNongracefulDisconnect (mysql) BinaryLogClient keepalive 스레드가 오래된(stale) 연결의 연결 끊기를 트리거할 때 socket.setSoLinger(true, 0)를 쓸지 여부예요. false boolean

엔드포인트 옵션 (Endpoint Options)

Debezium MySQL Connector 엔드포인트는 URI 문법으로 설정해요:

debezium-mysql:name

다음 _path_와 query 파라미터를 씁니다:

경로 파라미터 (Path Parameters)

Name Description Default Type
name (consumer) 필수 커넥터의 고유 이름이에요. 같은 이름으로 다시 등록하려 하면 실패해요. String

쿼리 파라미터 (Query Parameters)

Name Description Default Type
additionalProperties (common) debezium 컴포넌트 설정에 직접 담을 수 없는 추가 프로퍼티를 넣기 위한 옵션이에요(예: Debezium 엔진에 필요한 Kafka Connect 프로퍼티 설정, 예컨대 KafkaOffsetBackingStore 설정). 프로퍼티 앞에는 additionalProperties. 접두사를 붙여야 해요. 예: additionalProperties.transactional.id=12345&additionalProperties.schema.registry.url=http://localhost:8811/avro. additionalProperties. 접두사를 쓰는 다중 값(multi-value) 옵션이에요. Map
internalKeyConverter (consumer) 오프셋의 키 데이터를 직렬화·역직렬화하는 데 쓸 Converter 클래스예요. 기본값은 JSON converter예요. org.apache.kafka.connect.json.JsonConverter String
internalValueConverter (consumer) 오프셋의 값 데이터를 직렬화·역직렬화하는 데 쓸 Converter 클래스예요. 기본값은 JSON converter예요. org.apache.kafka.connect.json.JsonConverter String
offsetCommitPolicy (consumer) 커밋 정책(commit policy)의 Java 클래스 이름이에요. 처리한 이벤트 수와 마지막 커밋 이후 경과 시간을 기준으로 오프셋 커밋을 언제 트리거할지 정의해요. 이 클래스는 'OffsetCommitPolicy' 인터페이스를 구현해야 하고, 기본값은 시간 간격 기반의 주기적 커밋 정책이에요. String
offsetCommitTimeoutMs (consumer) 레코드를 플러시하고 파티션 오프셋 데이터를 오프셋 스토리지에 커밋하기까지 기다릴 최대 밀리초예요. 이 시간을 넘으면 프로세스를 취소하고, 해당 오프셋 데이터 커밋은 나중에 다시 시도하도록 복원해요. 기본값은 5초입니다. 5000 long
offsetFlushIntervalMs (consumer) 오프셋 커밋을 시도하는 간격이에요. 기본값은 1분입니다. 60000 long
offsetStorage (consumer) 커넥터 오프셋을 영속화하는 일을 맡는 Java 클래스 이름이에요. org.apache.kafka.connect.storage.FileOffsetBackingStore String
offsetStorageFileName (consumer) 오프셋을 저장할 파일 경로예요. offset.storage가 FileOffsetBackingStore로 설정된 경우 필요해요. String
offsetStoragePartitions (consumer) 오프셋 스토리지 토픽을 만들 때 쓰는 파티션 수예요. offset.storage가 'KafkaOffsetBackingStore'로 설정된 경우 필요합니다. int
offsetStorageReplicationFactor (consumer) 오프셋 스토리지 토픽을 만들 때 쓰는 복제 계수(replication factor)예요. offset.storage가 KafkaOffsetBackingStore로 설정된 경우 필요합니다. int
offsetStorageTopic (consumer) 오프셋을 저장할 Kafka 토픽 이름이에요. offset.storage가 KafkaOffsetBackingStore로 설정된 경우 필요해요. String
bridgeErrorHandler (consumer (advanced)) 컨슈머를 Camel 라우팅 에러 핸들러(Error Handler)에 브리지할 수 있게 해 줘요. 즉, Camel 컨슈머가 들어오는 메시지를 집어 올리는 동안 등에서 예외가 발생하면(가능한 경우), 그 예외를 메시지로 처리해서 라우팅 에러 핸들러가 다루게 돼요. 중요: 이것은 서드파티 컴포넌트가 예외가 던져졌을 때 Camel에 알릴 수 있도록 허용하는 경우에만 가능해요. 일부 컴포넌트는 내부에서만 처리하기 때문에 bridgeErrorHandler를 쓸 수 없어요. 다른 상황에서는 Camel 컴포넌트를 개선해서 서드파티 컴포넌트에 연결(hook)해 이를 가능하게 만들 수도 있어요. 기본으로 컨슈머는 org.apache.camel.spi.ExceptionHandler로 예외를 처리하는데, WARN 또는 ERROR 레벨로 로그를 남기고 무시해요. false boolean
exceptionHandler (consumer (advanced)) 컨슈머가 커스텀 ExceptionHandler를 쓰게 할 수 있어요. bridgeErrorHandler 옵션이 활성화되어 있으면 이 옵션은 사용되지 않는다는 점에 주의하세요. 기본으로 컨슈머는 예외를 처리하는데, WARN 또는 ERROR 레벨로 로그를 남기고 무시해요. ExceptionHandler
exchangePattern (consumer (advanced))
컨슈머가 교환(exchange)을 만들 때 교환 패턴을 설정해요.

Enum values:

  • InOnly

  • InOut

| | ExchangePattern | | bigintUnsignedHandlingMode (mysql) | BIGINT UNSIGNED 컬럼을 변경 이벤트에 어떻게 표현할지 정해요. 'precise'는 java.math.BigDecimal로 값을 표현하고, 변경 이벤트에는 이진 표현과 Kafka Connect의 'org.apache.kafka.connect.data.Decimal' 타입으로 인코딩돼요. 'long'(기본값)은 Java의 'long'으로 값을 표현하는데, 정밀도는 떨어질 수 있지만 consumer에서 훨씬 다루기 쉬워요. | long | String | | binlogBufferSize (mysql) | binlog reader가 진행 중인 트랜잭션이 커밋될지 롤백될지 판단하려고 쓰는 룩어헤드(look-ahead) 버퍼 크기예요. 0으로 두면 룩어헤드 버퍼링을 끕니다. 기본값은 0(버퍼링 꺼짐)이에요. | 0 | int | | binlogNetReadTimeout (mysql) | binlog 연결에서 읽기가 끝나기를 서버가 타임아웃 전까지 기다리는 초예요. 0이면 MySQL 서버 기본값을 씁니다. 스트리밍 중 EOFException을 막으려면 지연이 큰 네트워크 환경에서 이 값을 늘려야 할 수 있어요. | 0 | long | | binlogNetWriteTimeout (mysql) | binlog 연결에 쓰기가 끝나기를 서버가 타임아웃 전까지 기다리는 초예요. 0이면 MySQL 서버 기본값을 씁니다. 대용량 데이터로 스트리밍 중 EOFException이 나면 이 값을 늘려야 할 수 있어요. | 0 | long | | columnExcludeList (mysql) | 변경 이벤트에서 제외할 컬럼과 맞아떨어지는 정규식이에요. | | String | | columnIncludeList (mysql) | 변경 이벤트에 포함할 컬럼과 맞아떨어지는 정규식이에요. | | String | | columnPropagateSourceType (mysql) | 컬럼의 완전한 이름(fully-qualified name)과 맞는 정규식을 콤마로 나열한 값이에요. 이에 맞는 컬럼은 원래 타입과 원래 길이를 파라미터로 내보내는 변경 레코드의 필드 스키마에 추가돼요. | | String | | connectionValidationTimeoutMs (mysql) | 연결 검증이 끝나기를 기다리는 최대 시간(밀리초)이에요. 기본값은 60초입니다. | 1m | long | | connectKeepAlive (mysql) | 연결이 살아 있도록 유지하려고 별도 스레드를 쓸지 여부예요. | true | boolean | | connectKeepAliveIntervalMs (mysql) | keep alive 스레드를 쓸 때 연결을 확인하는 간격(밀리초)이에요. 기본값은 1분(60,000 ms)입니다. | 1m | long | | connectTimeoutMs (mysql) | 데이터베이스에 연결을 시도한 뒤 타임아웃되기까지 기다리는 최대 시간(밀리초)이에요. 기본값은 30초(30,000 ms)입니다. | 30s | int | | converters (mysql) | 기본 컨버터 대신 쓸 커스텀 컨버터 목록(선택)이에요. 컨버터는 '.type' 설정 옵션으로 정의하고 '.' 옵션으로 구성합니다. | | String | | customMetricTags (mysql) | 커스텀 메트릭 태그는 키-값 쌍을 받아서 MBean 객체 이름을 커스터마이즈해요. 이 이름은 일반 이름 끝에 붙고, 각 키는 MBean 객체 이름의 태그가 되며 그 값은 해당 태그의 값이 됩니다. 예: k1=v1,k2=v2. | | String | | customSanitizePattern (mysql) | 값을 마스킹할 설정 키를 골라내는 정규식이에요. 설정하면 Debezium의 기본 비밀번호 마스킹 패턴 대신 이 커스텀 패턴을 씁니다. | .*secret$|.*password$|.*sasl\.jaas\.config$|.*basic\.auth\.user\.info|.*registry\.auth\.client-secret|.*credentials\.json$|.*connectionstring$|.*connection\.string$ | String | | databaseExcludeList (mysql) | 모니터링에서 제외할 데이터베이스 이름과 맞는 정규식을 콤마로 나열한 값이에요. | | String | | databaseHostname (mysql) | 데이터베이스 서버의 DNS로 풀리는 호스트 이름이나 IP 주소예요. | | String | | databaseIncludeList (mysql) | 변경을 캡처할 데이터베이스들이에요. | | String | | databaseInitialStatements (mysql) | 데이터베이스에 JDBC 연결(binlog 읽기 연결 제외)이 맺어질 때 실행할 SQL 문을 세미콜론으로 나열한 값이에요. 커넥터가 임의로 JDBC 연결을 맺을 수 있다는 점에 유의하세요. 그래서 보통은 DML 문 실행이 아니라 세션 파라미터 설정 용도로만 쓰는 게 좋습니다. 세미콜론을 구분자가 아닌 문자로 쓰려면 (';;')처럼 두 개를 겹쳐 쓰세요. | | String | | databaseJdbcDriver (mysql) | MySQL 데이터베이스 서버에 연결할 때 쓰는 JDBC 드라이버 클래스 이름이에요. | com.mysql.cj.jdbc.Driver | String | | databasePassword (mysql) | 필수 데이터베이스에 연결할 때 쓸 데이터베이스 사용자의 비밀번호예요. | | String | | databasePort (mysql) | 데이터베이스 서버의 포트예요. | 3306 | int | | databaseProtocol (mysql) | 드라이버와 함께 쓸 JDBC 프로토콜이에요. | jdbc:mysql | String | | databaseQueryTimeoutMs (mysql) | 쿼리 실행을 기다리는 시간(밀리초)이에요. 기본값은 600초(600,000 ms)이고, 0이면 제한이 없어요. | 10m | int | | databaseServerId (mysql) | 이 데이터베이스 클라이언트의 숫자 ID예요. 클러스터에서 현재 도는 모든 데이터베이스 프로세스 사이에서 유일해야 합니다. 커넥터는 이 고유 ID를 가진 또 하나의 서버로서 데이터베이스 클러스터에 합류해서 binlog를 읽어요. | | long | | databaseServerIdOffset (mysql) | 병렬 스냅샷(parallel snapshotting)을 설정한 경우에만 관련 있어요. 병렬 스냅샷 중에는 데이터베이스 클라이언트에 여러(4개) 연결이 열리고, 각 연결은 고유한 연결 ID가 필요해요. 이 오프셋은 기준이 되는 클러스터 ID로부터 그 ID들을 만들어 낼 때 씁니다. | 10000 | long | | databaseSslKeystore (mysql) | 키 스토어 파일의 위치예요. 선택 사항이고, 클라이언트와 데이터베이스 사이의 양방향 인증(two-way authentication)에 쓸 수 있어요. | | String | | databaseSslKeystorePassword (mysql) | 키 스토어 파일의 비밀번호예요. 선택 사항이고, 'database.ssl.keystore'를 설정한 경우에만 필요합니다. | | String | | databaseSslMode (mysql) | 데이터베이스에 암호화 연결을 쓸지 정해요. 옵션: 'disabled'는 암호화하지 않은 연결을 써요. 'preferred'(기본값)는 서버가 보안 연결을 지원하면 보안(암호화) 연결을 맺고, 아니면 암호화하지 않은 연결로 물러나요. 'required'는 보안(암호화) 연결을 쓰되, 맺을 수 없으면 실패해요. 'verify_ca'는 'required'에 더해 설정된 인증 기관(CA) 인증서로 서버 TLS 인증서를 검증하고, 맞는 유효한 CA 인증서가 없으면 실패해요. 'verify_identity'는 'verify_ca'에 더해 서버 인증서가 접속하려는 호스트와 맞는지도 검증합니다. | preferred | String | | databaseSslTruststore (mysql) | 서버 인증서 검증에 쓸 트러스트 스토어 파일의 위치예요. | | String | | databaseSslTruststorePassword (mysql) | 트러스트 스토어 파일의 비밀번호예요. 트러스트스토어 무결성 검사와 잠금 해제에 씁니다. | | String | | databaseUser (mysql) | 데이터베이스에 연결할 때 쓸 데이터베이스 사용자 이름이에요. | | String | | datatypePropagateSourceType (mysql) | 데이터베이스별 데이터 타입 이름과 맞는 정규식을 콤마로 나열한 값이에요. 이에 맞는 데이터 타입은 원래 타입과 원래 길이를 파라미터로 내보내는 변경 레코드의 필드 스키마에 추가돼요. | | String | | ddlParserType (mysql) | MySQL DDL 문을 파싱할 때 어떤 ANTLR 문법을 쓸지 정해요. 'default'는 Oracle MySQL 문법을 쓰는데, 꾸준히 관리되고 MySQL 8.0 기능을 지원해요. 'legacy'는 기존 배포 환경과의 하위 호환을 위해 Positive Technologies 문법을 씁니다. | default | String | | decimalHandlingMode (mysql) | DECIMAL과 NUMERIC 컬럼을 변경 이벤트에 어떻게 표현할지 정해요. 'precise'(기본값)는 java.math.BigDecimal로 값을 표현하고, 변경 이벤트에는 이진 표현과 Kafka Connect의 'org.apache.kafka.connect.data.Decimal' 타입으로 인코딩돼요. 'string'은 문자열로 값을 표현해요. 'double'은 Java의 'double'로 값을 표현하는데, 정밀도는 떨어질 수 있지만 consumer에서 훨씬 다루기 쉬워요. | precise | String | | enableTimeAdjuster (mysql) | 데이터베이스는 연도 값을 2자리 또는 4자리로 넣게 허용해요. 2자리면 값이 1970~2069 범위로 자동 매핑돼요. false는 암시적 변환을 데이터베이스에 맡기고, true(기본값)는 Debezium이 변환을 수행해요. | true | boolean | | errorsMaxRetries (mysql) | 실패하기 전까지 연결 오류를 재시도할 최대 횟수예요(-1 = 제한 없음, 0 = 비활성화, 그 외 = 재시도 횟수). | -1 | int | | eventDeserializationFailureHandlingMode (mysql) | binlog 이벤트 역직렬화 중 장애(즉, 손상된 이벤트를 만났을 때)를 어떻게 처리할지 정해요. 'fail'(기본값)은 문제 이벤트와 그 binlog 위치를 알리는 예외를 던져 커넥터를 멈춰요. 'warn'은 문제 이벤트와 binlog 위치를 로그로 남기고 이벤트를 건너뛰어요. 'ignore'는 문제 이벤트를 조용히 건너뜁니다. | fail | String | | eventProcessingFailureHandlingMode (mysql) | 이벤트 처리 중 장애(즉, 손상된 이벤트를 만났을 때)를 어떻게 처리할지 정해요. 'fail'(기본값)은 문제 이벤트와 그 위치를 알리는 예외를 던져 커넥터를 멈춰요. 'warn'은 문제 이벤트와 위치를 로그로 남기고 이벤트를 건너뛰어요. 'ignore'는 문제 이벤트를 조용히 건너뜁니다. | fail | String | | executorShutdownTimeoutMs (mysql) | 태스크 익스큐터(task executor)가 꺼지기를 기다리는 최대 시간(밀리초)이에요. | 4s | long | | extendedHeadersEnabled (mysql) | 다운스트림 처리 시스템에서 CDC 이벤트의 출처를 추적·식별하는 데 꼭 필요한 메타데이터를 담은 Debezium 컨텍스트 헤더를 켜거나 꺼요. | true | boolean | | gtidIgnoreOnRecovery (mysql) | 복구 중에 GTID를 무시하고 대신 binlog 파일과 위치부터 다시 시작할지 여부예요. 서버의 GTID 모드는 그대로 켜져 있고, 복구 후에는 GTID 추적이 정상으로 돌아옵니다. | false | boolean | | gtidSourceExcludes (mysql) | MySQL 서버 binlog에서 시작 위치를 정할 때 GTID 범위를 제외하는 데 쓰는 소스 UUID들이에요. | | String | | gtidSourceFilterDmlEvents (mysql) | true로 두면, gtid.source.includes나 gtid.source.excludes(설정한 경우)가 정의한 GTID와 맞는 트랜잭션만 골라서 DML 이벤트를 만들어요. | true | boolean | | gtidSourceIncludes (mysql) | MySQL 서버 binlog에서 시작 위치를 정할 때 GTID 범위를 포함하는 데 쓰는 소스 UUID들이에요. | | String | | guardrailCollectionsLimitAction (mysql) | guardrail 컬렉션 한도를 넘었을 때 취할 동작을 정해요. 'warn'(기본값)은 경고 메시지를 로그로 남기고 처리를 계속하고, 'fail'은 에러로 커넥터를 멈춥니다. | warn | String | | guardrailCollectionsMax (mysql) | 커넥터가 캡처할 수 있는 컬렉션·테이블의 최대 개수예요. 한도를 넘으면 'guardrail.collections.limit.action'이 정한 동작을 합니다. 이 guardrail을 끄려면 0으로 두세요. | 0 | int | | heartbeatActionQuery (mysql) | 하트비트마다 실행할 쿼리예요. | | String | | heartbeatIntervalMs (mysql) | 커넥터가 하트비트 토픽에 주기적으로 하트비트 메시지를 보내는 간격(밀리초)이에요. 0으로 두면 하트비트 메시지를 끕니다. 기본은 꺼져 있어요. | 0ms | int | | heartbeatTopicsPrefix (mysql) | 하트비트 토픽 이름에 쓰는 접두사예요. 기본값은 __debezium-heartbeat입니다. | __debezium-heartbeat | String | | includeQuery (mysql) | 변경 이벤트를 만든 원본 SQL 쿼리를 이벤트에 포함할지 여부예요. 참고: 이 옵션을 쓰려면 데이터베이스가 서버 옵션 binlog_rows_query_log_events(MySQL) 또는 binlog_annotate_row_events(MariaDB)를 ON으로 설정된 상태여야 해요. 스냅샷에서 만든 이벤트에는 쿼리가 담기지 않아요. 경고: 이 옵션을 켜면 변경 이벤트에 원본 SQL 문이 담기면서, 의도적으로 제외하거나 마스킹한 테이블·필드가 노출될 수 있어요. 그래서 기본값은 'false'입니다. | false | boolean | | includeSchemaChanges (mysql) | 데이터베이스 스키마 변경을 데이터베이스 서버 ID와 같은 이름의 Kafka 토픽에 발행할지 여부예요. 각 스키마 변경은 데이터베이스 이름을 담은 키로 기록되고, 값에는 새 스키마의 논리적 설명과 (있다면) DDL 문이 들어가요. 기본값은 'true'입니다. 이는 커넥터가 내부적으로 데이터베이스 스키마 히스토리를 기록하는 방식과는 별개예요. | true | boolean | | includeSchemaComments (mysql) | 테이블·컬럼 주석을 메타데이터 객체로 파싱할지 여부예요. 참고: 이 옵션을 켜면 메모리 사용량에 영향이 있어요. Debezium 커넥터의 메모리 소비는 ColumnImpl 객체의 개수와 크기가 좌우하는데, 객체마다 String을 하나씩 붙이면 꽤 무거워질 수 있습니다. 기본값은 'false'예요. | false | boolean | | inconsistentSchemaHandlingMode (mysql) | 내부 스키마 표현에 없는 테이블(즉, 내부 표현이 데이터베이스와 일치하지 않음)에 속한 binlog 이벤트를 어떻게 처리할지 정해요. 'fail'(기본값)은 문제 이벤트와 그 binlog 위치를 알리는 예외를 던져 커넥터를 멈춰요. 'warn'은 문제 이벤트와 binlog 위치를 로그로 남기고 이벤트를 건너뛰어요. 'skip'은 문제 이벤트를 조용히 건너뜁니다. | fail | String | | incrementalSnapshotAllowSchemaChanges (mysql) | 증분 스냅샷 중 스키마 변경을 감지하고 현재 청크를 다시 골라서 잠금(locking) DDL을 피하게 해 줘요. 기본 키 변경은 지원하지 않고, 증분 스냅샷 중에 수행하면 잘못된 결과를 낼 수 있다는 점에 유의하세요. 또한 스키마 변경이 컬럼 기본값에만 영향을 준다면, binlog 스트림에서 DDL을 처리하기 전까지는 변경을 감지하지 못해요. 이는 스냅샷 이벤트의 값에는 영향을 주지 않지만, 스냅샷 이벤트의 스키마에는 오래된 기본값이 남을 수 있습니다. | false | boolean | | incrementalSnapshotChunkSize (mysql) | 증분 스냅샷에서 청크(문서/행 개수)의 최대 크기예요. | 1024 | int | | incrementalSnapshotWatermarkingStrategy (mysql) | 증분 스냅샷 동안 워터마킹(watermarking)에 쓸 전략을 정해요. 'insert_insert'는 열기(open)와 닫기(close) 시그널을 모두 시그널 데이터 컬렉션에 기록해요(기본값). 'insert_delete'는 열기 시그널만 시그널 데이터 컬렉션에 기록하고, 닫기가 상응하는 열기 시그널을 지웁니다. | INSERT_INSERT | String | | legacySnapshotMaxThreads (mysql) | 켜면 구 방식(table-per-thread) 병렬 스냅샷 알고리즘을 써요. false(기본값)로 두면 테이블을 청크로 쪼개 모든 스냅샷 스레드에 나눠 처리해서, 스냅샷 동시성을 더 높일 수 있어요. | false | boolean | | maxBatchSize (mysql) | 소스 레코드 배치 하나의 최대 크기예요. 기본값은 2048입니다. | 2048 | int | | maxQueueSize (mysql) | 데이터베이스 로그에서 읽었지만 아직 기록·전달되지 않은 변경 이벤트 큐의 최대 크기예요. 기본값은 8192이고, 항상 최대 배치 크기보다 커야 합니다. | 8192 | int | | maxQueueSizeInBytes (mysql) | 데이터베이스 로그에서 읽었지만 아직 기록·전달되지 않은 변경 이벤트 큐의 최대 크기(바이트)예요. 기본값은 0이고, 이는 기능이 꺼져 있다는 뜻이에요. | 0 | long | | memoryManagementSchemasClass (mysql) | 스키마 메타데이터 저장 구현체의 완전한 클래스 이름(fully-qualified)이에요. 이 클래스는 io.debezium.relational.TableMappingStorage를 구현해야 합니다. 기본값은 인메모리 저장용 io.debezium.relational.ConcurrentMapTableMappingStorage예요. | io.debezium.relational.ConcurrentMapTableMappingStorage | String | | memoryManagementTablesClass (mysql) | 테이블 메타데이터 저장 구현체의 완전한 클래스 이름(fully-qualified)이에요. 이 클래스는 io.debezium.relational.TableMappingStorage를 구현해야 합니다. 기본값은 인메모리 저장용 io.debezium.relational.ConcurrentMapTableMappingStorage예요. | io.debezium.relational.ConcurrentMapTableMappingStorage | String | | messageKeyColumns (mysql) | 메시지 키로 쓸 완전한 이름(fully-qualified)의 테이블·컬럼과 맞는 표현식을 세미콜론으로 나열한 값이에요. 각 표현식은 ':' 패턴과 일치해야 하고, 테이블 이름은 커넥터에 따라 (DB_NAME.TABLE_NAME) 또는 (SCHEMA_NAME.TABLE_NAME)으로 정의할 수 있어요. 키 컬럼은 커스텀 키가 되는 컬럼들을 콤마로 나열한 값입니다. 명시적으로 키 설정이 없는 테이블은 그 테이블의 기본 키 컬럼이 메시지 키로 쓰여요. 예: dbserver1.inventory.orderlines:orderId,orderLineId;dbserver1.inventory.orders:id. | | String | | minRowCountToStreamResults (mysql) | 스냅샷 동안 결과를 전부 메모리로 당겨 오는 대신 스트리밍하려면, 테이블이 가져야 하는 최소 행 수예요. 기본값은 1,000입니다. 0으로 두면 모든 결과를 스트리밍해서 테이블 크기 확인을 아예 생략해요. | 1000 | int | | notificationEnabledChannels (mysql) | 활성화할 알림(notifications) 채널 이름 목록이에요. | | String | | notificationSinkTopicName (mysql) | 알림을 보낼 토픽 이름이에요. 활성 채널 목록에 'sink'가 있으면 필수입니다. | | String | | openlineageIntegrationConfigFilePath (mysql) | OpenLineage 설정 파일의 경로예요. https://openlineage.io/docs/client/java/configuration을 참고하세요. | ./openlineage.yml | String | | openlineageIntegrationDatasetKafkaBootstrapServers (mysql) | 입력/출력 네임스페이스로 쓰는 Kafka 부트스트랩 서버 주소예요. | | String | | openlineageIntegrationEnabled (mysql) | Debezium이 OpenLineage API로 데이터 계보(data lineage) 메타데이터를 내보내도록 켜요. | false | boolean | | openlineageIntegrationJobDescription (mysql) | Debezium이 내보내는 잡(job) 설명이에요. | Debezium change data capture job | String | | openlineageIntegrationJobNamespace (mysql) | Debezium이 내보내는 잡 네임스페이스예요. | | String | | openlineageIntegrationJobOwners (mysql) | Debezium이 내보내는 잡 소유자예요. 키-값 쌍을 콤마로 나열합니다. 예: k1=v1,k2=v2. | | String | | openlineageIntegrationJobTags (mysql) | Debezium이 내보내는 잡 태그예요. 키-값 쌍을 콤마로 나열합니다. 예: k1=v1,k2=v2. | | String | | pollIntervalMs (mysql) | 이벤트가 오지 않은 뒤 새 변경 이벤트를 기다리는 시간(밀리초)이에요. 기본값은 500 ms입니다. | 500ms | long | | postProcessors (mysql) | 포스트 프로세서 목록(선택)이에요. 프로세서는 '.type' 설정 옵션으로 정의하고 '' 옵션으로 구성합니다. | | String | | provideTransactionMetadata (mysql) | 이벤트 카운트와 함께 트랜잭션 메타데이터 추출을 켜요. | false | boolean | | queryFetchSize (mysql) | 스트리밍 동안 메모리로 가져올 레코드의 최대 개수예요. '0'이면 기본 JDBC fetch size를 씁니다. | 0 | int | | retriableRestartConnectorWaitMs (mysql) | 재시도 가능한 예외가 발생한 뒤 커넥터를 다시 시작하기 전에 기다리는 시간이에요. 기본값은 10000ms입니다. | 10s | long | | schemaHistoryInternal (mysql) | 데이터베이스 스키마 변경을 저장·복구하는 데 쓸 SchemaHistory 클래스 이름이에요. 히스토리 관련 설정 프로퍼티는 'schema.history.internal.' 문자열을 접두사로 씁니다. | io.debezium.storage.kafka.history.KafkaSchemaHistory | String | | schemaHistoryInternalFileFilename (mysql) | 데이터베이스 스키마 히스토리를 기록할 파일의 경로예요. | | String | | schemaHistoryInternalSkipUnparseableDdl (mysql) | binlog에서 파싱할 수 없는 DDL 문을 만났을 때 Debezium이 취할 동작을 정해요. 기본으로 커넥터는 동작을 멈추지만, 설정을 바꾸면 파싱할 수 없는 문을 무시할 수 있어요. 건너뛰기를 켜면 Debezium이 메타데이터 변경을 놓칠 수 있습니다. | false | boolean | | schemaHistoryInternalStoreOnlyCapturedDatabasesDdl (mysql) | Debezium이 데이터베이스 스키마 히스토리에 어떤 DDL을 저장할지 정해요. 기본(false)은 들어오는 모든 DDL 문을 저장해요. true로 두면 캡처한 스키마/데이터베이스의 테이블을 건드리는 DDL만 저장합니다. | false | boolean | | schemaHistoryInternalStoreOnlyCapturedTablesDdl (mysql) | Debezium이 데이터베이스 스키마 히스토리에 어떤 DDL을 저장할지 정해요. 기본(false)은 들어오는 모든 DDL 문을 저장해요. true로 두면 캡처한 테이블을 건드리는 DDL만 저장합니다. | false | boolean | | schemaNameAdjustmentMode (mysql) | 커넥터가 쓰는 메시지 컨버터와 호환되도록 스키마 이름을 어떻게 조정할지 정해요. 'avro'는 Avro 타입 이름에 쓸 수 없는 문자를 밑줄(underscore)로 바꿔요. 'avro_unicode'는 Avro 타입 이름에 쓸 수 없는 문자나 밑줄을 _uxxxx 같은 해당 유니코드로 바꿉니다. 참고: _는 Java의 백슬래시 같은 이스케이프 시퀀스예요. 'none'은 아무 조정도 하지 않아요(기본값). | none | String | | signalDataCollection (mysql) | Debezium에 시그널/명령을 보낼 때 쓰는 데이터 컬렉션 이름이에요. 멀티 파티션 모드 커넥터는 시그널 데이터 컬렉션을 콤마로 구분한 목록으로 여러 개 지정할 수 있어요. 설정하지 않으면 시그널링이 꺼집니다. | | String | | signalEnabledChannels (mysql) | 활성화할 시그널(signal) 채널 이름 목록이에요. source 채널이 기본으로 켜져 있습니다. | source | String | | signalPollIntervalMs (mysql) | 등록된 채널에서 새 시그널을 찾는 간격(밀리초)이에요. 기본값은 5초입니다. | 5s | long | | skippedOperations (mysql) | 스트리밍 중 건너뛸 연산을 콤마로 나열한 값이에요. 'c'는 insert/create, 'u'는 update, 'd'는 delete, 't'는 truncate, 'none'은 건너뛰지 않음을 뜻해요. 기본으로는 truncate 연산만 건너뜁니다. | t | String | | snapshotDelayMs (mysql) | 스냅샷이 시작되기 전에 기다리는 지연 시간(밀리초)이에요. 기본값은 0 ms입니다. | 0ms | long | | snapshotFetchSize (mysql) | 스냅샷을 수행하는 동안 메모리로 가져올 레코드의 최대 개수예요. | | int | | snapshotIncludeCollectionList (mysql) | 커넥터를 만들거나 다시 시작할 때 스냅샷을 찍을 테이블/컬렉션 목록을 지정하려면 이 설정을 반드시 써야 해요. | | String | | snapshotLockingMode (mysql) | 스냅샷을 수행하는 동안 커넥터가 전역 읽기 잠금(global read lock)을 얼마나 오래 쥐고 있을지 정해요. 기본값인 'minimal'은 데이터베이스 스키마 등 다른 메타데이터를 읽는 스냅샷 초기 구간 동안만 전역 읽기 잠금을 쥐고(그래서 업데이트를 막고) 있어요. 스냅샷의 나머지 작업은 각 테이블에서 모든 행을 읽는 건데, 잠금을 놓은 뒤에도 다른 연산이 데이터베이스를 갱신하는 동안 스냅샷 프로세스의 REPEATABLE READ 트랜잭션으로 처리할 수 있어요. 하지만 어떤 경우에는 스냅샷 내내 모든 쓰기를 막는 게 좋을 수도 있어요. 그럴 때는 이 프로퍼티를 'extended'로 두세요. 'none'으로 두면 스냅샷 과정에서 커넥터가 어떤 테이블 잠금도 걸지 않아요. 이 모드는 snapshot.mode가 'schema_only' 또는 'schema_only_recovery'일 때만 조합할 수 있고, 스냅샷을 찍는 동안 스키마 변경이 일어나지 않을 때만 안전합니다. | minimal | String | | snapshotLockTimeoutMs (mysql) | 스냅샷 시작 시 테이블 잠금을 기다리는 최대 밀리초예요. 이 시간 안에 잠금을 얻지 못하면 스냅샷이 중단돼요. 기본값은 10초입니다. | 10s | long | | snapshotMaxThreads (mysql) | 스냅샷을 수행하는 데 쓰는 스레드의 최대 개수예요. 기본값은 1입니다. | 1 | int | | snapshotMaxThreadsMultiplier (mysql) | 테이블별 스냅샷 청크 개수를 조절하는 배율이에요. 기본 동작은 'row_count/snapshot.max.threads'로 청크당 행 수를 계산해요. 큰 테이블에서는 이게 최선이 아닐 수 있는데, 배율을 쓰면 'row_count/(snapshot.max.threads snapshot.max.threads.multiplier)' 공식으로 청크 수를 늘립니다. | 1 | int | | snapshotMode (mysql) | 커넥터 시작 시 스냅샷을 실행하는 기준이에요. 다음 스냅샷 옵션 중 하나를 고르세요. 'when_needed': 시작할 때 스냅샷이 필요하면 실행해요. 'schema_only': 논리적 서버 이름에 대한 오프셋이 없으면 테이블 구조(스키마)만 캡처하는 스냅샷을 실행하고, 테이블 데이터는 담지 않아요. 스냅샷이 끝나면 binlog에서 변경을 스트리밍하기 시작합니다. 'schema_only_recovery': 데이터베이스 스키마 히스토리만 캡처하는 스냅샷을 수행하고, 다시 스트리밍으로 돌아가요. 손상되거나 잃어버린 데이터베이스 스키마 히스토리 토픽을 복구할 때 쓰세요. 커넥터가 멈춘 뒤에 데이터베이스 스키마가 바뀌었다면 쓰지 마세요. 'initial'(기본값): 논리적 서버 이름에 대한 오프셋이 없으면 설정한 테이블의 현재 전체 상태를 캡처하는 스냅샷을 실행해요. 스냅샷이 끝나면 binlog에서 변경을 스트리밍하기 시작합니다. 'initial_only': 'initial'처럼 스냅샷을 수행하지만, 스냅샷이 끝나면 멈추고 binlog에서 변경을 스트리밍하지 않아요. | initial | String | | snapshotModeConfigurationBasedSnapshotData (mysql) | 'snapshot.mode'가 configuration_based로 설정됐을 때, 데이터를 스냅샷에 담을지 여부를 정하는 설정이에요. | false | boolean | | snapshotModeConfigurationBasedSnapshotOnDataError (mysql) | 'snapshot.mode'가 configuration_based로 설정됐을 때, 에러가 났을 경우 데이터를 스냅샷에 담을지 여부를 정하는 설정이에요. | false | boolean | | snapshotModeConfigurationBasedSnapshotOnSchemaError (mysql) | 'snapshot.mode'가 configuration_based로 설정됐을 때, 에러가 났을 경우 스키마를 스냅샷에 담을지 여부를 정하는 설정이에요. | false | boolean | | snapshotModeConfigurationBasedSnapshotSchema (mysql) | 'snapshot.mode'가 configuration_based로 설정됐을 때, 스키마를 스냅샷에 담을지 여부를 정하는 설정이에요. | false | boolean | | snapshotModeConfigurationBasedStartStream (mysql) | 'snapshot.mode'가 configuration_based로 설정됐을 때, 스냅샷 뒤에 스트리밍을 시작할지 여부를 정하는 설정이에요. | false | boolean | | snapshotModeCustomName (mysql) | 'snapshot.mode'가 custom으로 설정됐을 때는 반드시 설정해야 해요. 'name()' 메서드가 제공하는 커스텀 구현의 이름을 지정합니다. 구현체는 'Snapshotter' 인터페이스를 구현해야 하고, 앱이 부팅될 때마다 스냅샷을 찍을지 판단하려고 호출돼요. | | String | | snapshotQueryMode (mysql) | 스냅샷 동안 쓸 쿼리를 제어해요. | select_all | String | | snapshotQueryModeCustomName (mysql) | 'snapshot.query.mode'가 custom으로 설정됐을 때는 반드시 설정해야 해요. 'name()' 메서드가 제공하는 커스텀 구현의 이름을 지정합니다. 구현체는 'SnapshotterQuery' 인터페이스를 구현해야 하고, 스냅샷 동안 쿼리를 어떻게 만들지 결정하려고 호출돼요. | | String | | snapshotSelectStatementOverrides (mysql) | 이 프로퍼티는 완전한 이름(fully-qualified)의 테이블 (DB_NAME.TABLE_NAME) 또는 (SCHEMA_NAME.TABLE_NAME)을 콤마로 나열한 값이에요(커넥터에 따라 다름). 테이블별 select 문은 이어지는 설정 프로퍼티에서 지정해요. 각 테이블마다 하나씩, 'snapshot.select.statement.overrides.DB_NAME.TABLE_NAME' 또는 'snapshot.select.statement.overrides.SCHEMA_NAME.TABLE_NAME' 아이디로 식별합니다. 그 프로퍼티 값은 스냅샷 중 해당 테이블에서 데이터를 읽을 때 쓸 select 문이에요. 대용량 append-only 테이블이라면, 이전 스냅샷이 중단됐을 때 스냅샷을 (재개) 시작할 지점을 정하는 용도로 쓸 수 있어요. | | String | | snapshotTablesOrderByRowCount (mysql) | 초기 스냅샷에서 테이블을 처리하는 순서를 제어해요. descending은 행 수 내림차순으로, ascending은 행 수 오름차순으로 테이블을 정렬해요. disabled(기본값)는 행 수 정렬을 끕니다. | disabled | String | | sourceinfoStructMaker (mysql) | SourceInfo 스키마와 struct를 돌려주는 SourceInfoStructMaker 클래스 이름이에요. | io.debezium.connector.mysql.MySqlSourceInfoStructMaker | String | | statisticsMetricsEnabled (mysql) | 레코드 처리 지연 시간 같은 여러 통계와, 퀀타일(quantiles) 같은 파생 데이터 수집을 켜요. 기본으로 통계 수집은 켜져 있습니다. | true | boolean | | streamingDelayMs (mysql) | 스냅샷이 끝나고 스트리밍이 시작되기 전까지의 지연 시간(밀리초)이에요. 기본값은 0 ms입니다. | 0ms | long | | tableExcludeList (mysql) | 모니터링에서 제외할 테이블의 완전한 이름(fully-qualified)과 맞는 정규식을 콤마로 나열한 값이에요. | | String | | tableIgnoreBuiltin (mysql) | 기본 제공(built-in) 테이블을 무시할지 정하는 플래그예요. | true | boolean | | tableIncludeList (mysql) | 변경을 캡처할 테이블들이에요. | | String | | timePrecisionMode (mysql) | 시간, 날짜, 타임스탬프는 여러 정밀도로 표현할 수 있어요. 'adaptive_time_microseconds': 날짜·타임스탬프 값의 정밀도는 데이터베이스 컬럼의 정밀도를 따르지만, time 필드는 항상 마이크로초 정밀도를 씁니다. 'connect': 데이터베이스 컬럼 정밀도와 무관하게 항상 밀리초 정밀도로, Kafka Connect의 Time·Date·Timestamp 내장 표현으로 시간·날짜·타임스탬프 값을 표현해요. | adaptive_time_microseconds | String | | tombstonesOnDelete (mysql) | 삭제 연산을 delete 이벤트와 그 뒤의 tombstone 이벤트로 나타낼지(true), delete 이벤트만으로 나타낼지(false) 정해요. tombstone 이벤트를 내보내면(기본 동작) 소스 레코드가 지워진 뒤 Kafka가 해당 키와 관련된 이벤트를 전부 지울 수 있어요. | false | boolean | | topicNamingStrategy (mysql) | 데이터 변경, 스키마 변경, 트랜잭션, 하트비트 이벤트 등의 토픽 이름을 정하는 데 쓸 TopicNamingStrategy 클래스 이름이에요. | io.debezium.schema.SchemaTopicNamingStrategy | String | | topicPrefix (mysql) | 필수 변경을 캡처하는 해당 데이터베이스 서버/클러스터를 식별하고 네임스페이스를 제공하는 토픽 접두사예요. 이 커넥터가 내보내는 이벤트를 받는 모든 Kafka 토픽 이름의 접두사로 쓰이므로, 다른 모든 커넥터 사이에서 유일해야 합니다. 영숫자와 하이픈, 점, 밑줄만 받아들이세요. | | String | | transactionMetadataFactory (mysql) | 트랜잭션 컨텍스트와 트랜잭션 struct/스키마를 만드는 클래스예요. | io.debezium.pipeline.txmetadata.DefaultTransactionMetadataFactory | String | | useNongracefulDisconnect (mysql) | BinaryLogClient keepalive 스레드가 오래된(stale) 연결의 연결 끊기를 트리거할 때 socket.setSoLinger(true, 0)를 쓸지 여부예요. | false | boolean |

메시지 헤더 (Message Headers)

Debezium MySQL Connector 컴포넌트가 지원하는 메시지 헤더는 아래와 같아요:

Name Description Default Type
CamelDebeziumSourceMetadata (consumer) Constant: HEADER_SOURCE_METADATA 소스 이벤트에 대한 메타데이터예요. 테이블 이름, 데이터베이스 이름, 로그 위치 등이 들어가고, 자세한 내용은 Debezium 문서를 참고하세요. Map
CamelDebeziumIdentifier (consumer) Constant: HEADER_IDENTIFIER 커넥터의 식별자예요. 보통 {server-name}.{database-name}.{table-name} 형식입니다. String
CamelDebeziumKey (consumer) Constant: HEADER_KEY 이벤트의 키예요. 보통 테이블의 기본 키(Primary Key)입니다. Struct
CamelDebeziumOperation (consumer) Constant: HEADER_OPERATION 있다면 이벤트 연산의 타입이에요. 커넥터에서 쓰는 값은 create(또는 insert)가 c, update가 u, delete가 d, 그리고 read(초기 동기화의 경우)나 스냅샷 이벤트는 r이에요. String
CamelDebeziumTimestamp (consumer) Constant: HEADER_TIMESTAMP 있다면 커넥터가 이벤트를 처리한 시각(JVM의 시스템 클록 기준)이에요. Long
CamelDebeziumBefore (consumer) Constant: HEADER_BEFORE 있다면 이벤트가 발생하기 전의 행 상태를 담고 있어요. Struct
CamelDebeziumDdlSQL (consumer) Constant: HEADER_DDL_SQL 있다면 이벤트의 DDL SQL 텍스트예요. String

설정에 대해 더 알고 싶다면 아래 문서들을 참고하세요: https://debezium.io/documentation/reference/0.10/operations/embedded.html#engine-properties https://debezium.io/documentation/reference/0.10/connectors/mysql.html#connector-properties

사용법 (Usage)

메시지 본문 (Message body)

메시지 본문이 null이 아니라면(tombstone인 경우) 이벤트가 발생한 뒤의 행 상태를 담고 있어요. 형식은 둘 중 하나입니다:

  • Struct

  • Map(Struct에서 Map으로 바꿔 주는 내장 Type Converter를 쓰는 경우)

자세한 내용은 아래를 보세요.

예시 (Examples)

이벤트 소비 (Consuming events)

MySQL 커넥터에서 Debezium 이벤트를 들으려면, 이런 기본 라우트를 쓸 수 있어요.

from("debezium-mysql:dbz-test-1?offsetStorageFileName=/usr/offset-file-1.dat&databaseHostname=localhost&databaseUser=debezium&databasePassword=dbz&topicPrefix=my-app-connector&schemaHistoryInternalFileFilename=/usr/history-file-1.dat")
    .log("Event received from Debezium : ${body}")
    .log("    with this identifier ${headers.CamelDebeziumIdentifier}")
    .log("    with these source metadata ${headers.CamelDebeziumSourceMetadata}")
    .log("    the event occurred upon this operation '${headers.CamelDebeziumOperation}'")
    .log("    on this database '${headers.CamelDebeziumSourceMetadata[db]}' and this table '${headers.CamelDebeziumSourceMetadata[table]}'")
    .log("    with the key ${headers.CamelDebeziumKey}")
    .log("    the previous value is ${headers.CamelDebeziumBefore}")
    .log("    the ddl sql text is ${headers.CamelDebeziumDdlSQL}");

<route>
  <from uri="debezium-mysql:dbz-test-1?offsetStorageFileName=/usr/offset-file-1.dat&amp;databaseHostname=localhost&amp;databaseUser=debezium&amp;databasePassword=dbz&amp;topicPrefix=my-app-connector&amp;schemaHistoryInternalFileFilename=/usr/history-file-1.dat"/>
  <log message="Event received from Debezium : ${body}"/>
  <log message="    with this identifier ${headers.CamelDebeziumIdentifier}"/>
  <log message="    with these source metadata ${headers.CamelDebeziumSourceMetadata}"/>
  <log message="    the event occurred upon this operation '${headers.CamelDebeziumOperation}'"/>
  <log message="    on this database '${headers.CamelDebeziumSourceMetadata[db]}' and this table '${headers.CamelDebeziumSourceMetadata[table]}'"/>
  <log message="    with the key ${headers.CamelDebeziumKey}"/>
  <log message="    the previous value is ${headers.CamelDebeziumBefore}"/>
  <log message="    the ddl sql text is ${headers.CamelDebeziumDdlSQL}"/>
</route>

- route:
    from:
      uri: debezium-mysql:dbz-test-1
      parameters:
        offsetStorageFileName: /usr/offset-file-1.dat
        databaseHostname: localhost
        databaseUser: debezium
        databasePassword: dbz
        topicPrefix: my-app-connector
        schemaHistoryInternalFileFilename: /usr/history-file-1.dat
      steps:
        - log:
            message: "Event received from Debezium : ${body}"
        - log:
            message: "    with this identifier ${headers.CamelDebeziumIdentifier}"
        - log:
            message: "    with these source metadata ${headers.CamelDebeziumSourceMetadata}"
        - log:
            message: "    the event occurred upon this operation '${headers.CamelDebeziumOperation}'"
        - log:
            message: "    on this database '${headers.CamelDebeziumSourceMetadata[db]}' and this table '${headers.CamelDebeziumSourceMetadata[table]}'"
        - log:
            message: "    with the key ${headers.CamelDebeziumKey}"
        - log:
            message: "    the previous value is ${headers.CamelDebeziumBefore}"
        - log:
            message: "    the ddl sql text is ${headers.CamelDebeziumDdlSQL}"

기본으로 컴포넌트는 이벤트를 본문과 CamelDebeziumBefore 헤더에 Struct 데이터 타입으로 담아 내보내요. 나중에 필요할 때 스키마 정보를 활용할 수 있도록 하기 위해서예요. 하지만 컴포넌트에는 기본 출력 타입인 Struct를 Map으로 바꿔 주는 Type Converter도 들어 있어서, Camel의 풍부한 Data Format 타입을 활용할 수 있습니다. 이 중 상당수는 Map 데이터 타입과 바로 잘 동작해요. 쓰는 방법은 두 가지예요: 메시지에 접근할 때 Map.class 타입을 지정하거나(예: exchange.getIn().getBody(Map.class)), 라우트 빌더에서 from 문 뒤에 .convertBodyTo(Map.class)를 붙여 본문을 항상 Map으로 변환하는 방법입니다.

위에서 스키마 이야기를 했는데, 고급 데이터 변환을 하려고 할 때 이 스키마가 필요해질 수 있어요. 본문을 Map으로 바꾸지 않기로 했다면, Struct에서 Schema 타입으로 스키마 정보를 이렇게 얻을 수 있습니다:

Java 전용: 프로세서에서 Struct 스키마에 접근하기

from("debezium-mysql:[name]?[options]])
    .process(exchange -> {
        final Struct bodyValue = exchange.getIn().getBody(Struct.class);
        final Schema schemaValue = bodyValue.schema();

        log.info("Body value is : {}", bodyValue);
        log.info("With Schema : {}", schemaValue);
        log.info("And fields of : {}", schemaValue.fields());
        log.info("Field name has `{}` type", schemaValue.field("name").schema());
    });

중요 앞서 말했듯 이 컴포넌트는 Debezium Engine을 감싸는 얇은 wrapper예요. 따라서 프로덕션에서 쓰기 전에 Debezium이 어떻게 동작하는지, 설정이 어떻게 기대한 동작으로 이어지는지 이해해야 해요. 특히 장애 처리 부분은 꼭 짚고 넘어가세요.

더 알아보기 (Learn more)

함께 읽는 실무 이야기

우리 조직에도 적용하고 싶다면, AI 도입·업무 자동화를 살펴보세요.