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

Debezium PostgreSQL 커넥터

원문 보기 위키 갱신

Debezium PostgreSQL 커넥터 (camel-debezium-postgres)

Debezium PostgreSQL 컴포넌트는 Kafka나 Kafka Connect 없이도 Debezium Engine으로 PostgreSQL 데이터베이스의 Change Data Capture(CDC) 이벤트를 라우트로 흘려 보내는 컴포넌트예요. 별도의 Kafka Connect 인프라를 띄우지 않고 애플리케이션 안에 임베디드 엔진으로 붙여 쓰는 방식이죠. 컨슈머 전용이라서 from()으로 CDC 이벤트를 구독하고, 오프셋 저장, 스냅샷, 복제 슬롯, 헬스 체크 관련 옵션들이 제공돼요.

출처: 문서

본문

Camel 3.0부터 지원

컨슈머만 지원해요

Debezium PostgreSQL 컴포넌트는 Debezium을 Debezium Engine으로 감싼 래퍼(wrapper)예요. Kafka나 Kafka Connect 없이도 Debezium으로 PostgreSQL 데이터베이스에서 Change Data Capture를 할 수 있게 해 줘요.

참고 장애 처리 관련 참고: Debezium Embedded Engine 문서에 따르면, 엔진은 소스 오프셋을 적극적으로 기록하고 이 오프셋을 주기적으로 영구 저장소에 플러시해요. 그래서 애플리케이션을 재시작하거나 크래시가 나도 엔진은 마지막으로 기록된 오프셋부터 이어서 처리해요. 정상 동작 중이라면 다운스트림 라우트는 각 이벤트를 정확히 한 번씩 받아요. 하지만 애플리케이션이 크래시로 죽으면(우아한 종료 없이) 마지막으로 기록된 오프셋부터 이어서 처리하기 때문에, 재시작 직후에는 이벤트가 중복해서 도착할 수 있어요. 그래서 다운스트림 라우트는 이런 경우를 견딜 수 있어야 하고, 필요하면 이벤트를 중복 제거(deduplicate)해야 해요.

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

임베디드 엔진이 오류로 멈추면, 예를 들어 커넥터가 데이터베이스에 도달할 수 없거나 오프셋 스토어를 읽을 수 없는 경우에, 엔진은 재시작되지 않아요: 라우트는 시작 상태로 남지만 더 이상 이벤트를 받지 못해요. 이 실패는 consumer exception handler로 보고되고 라우트 헬스 체크를 DOWN으로 바꿔서 readiness probe가 잡을 수 있게 해 줘요.

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

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

URI 형식 (URI Format)

debezium-postgres:name[?options]

옵션 설정 (Configuring Options)

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

  • 컴포넌트(component) 레벨

  • 엔드포인트(endpoint) 레벨

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

컴포넌트 레벨에서는 엔드포인트들이 이어서 물려받는 일반적이고 공유되는 설정을 해요. 가장 상위의 설정 레벨이죠.

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

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

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

  • Component DSL.

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

  • Java 코드에서 직접.

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

엔드포인트는 옵션이 많아서 설정에 시간을 더 쓰게 될 거예요. 이 옵션들이 엔드포인트가 원하는 일을 하도록 커스터마이즈해 줘요. 옵션은 엔드포인트가 컨슈머(from)로 쓰이는지, 프로듀서(to)로 쓰이는지, 둘 다인지에 따라 분류되기도 해요.

엔드포인트 설정은 대부분 엔드포인트 URI에 _path_와 query 파라미터로 바로 하는 경우가 많아요. Endpoint DSL과 DataFormat DSL을 쓰면 Java에서 엔드포인트와 데이터 포맷을 _타입 세이프_하게 설정할 수도 있어요.

옵션을 설정할 때 좋은 습관은 Property Placeholders를 사용하는 거예요.

프로퍼티 플레이스홀더는 이런 장점이 있어요:

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

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

  • 코드가 더 유연해지고 재사용하기 좋아져요.

다음 두 섹션에서 모든 옵션을 나열해요. 먼저 컴포넌트 옵션, 그다음 엔드포인트 옵션 순서예요.

컴포넌트 옵션 (Component Options)

Debezium PostgreSQL Connector 컴포넌트는 아래 나열된 옵션들을 지원해요.

Name Description Default Type
additionalProperties (common) camel 설정에 바로 넣을 수 없는 debezium 컴포넌트용 추가 프로퍼티예요(예: Debezium 엔진에 필요한 Kafka Connect 프로퍼티 설정, KafkaOffsetBackingStore 설정 같은 것). 프로퍼티에는 additionalProperties. 접두어를 붙여야 해요. 예: additionalProperties.transactional.id=12345&additionalProperties.schema.registry.url=http://localhost:8811/avro. additionalProperties. 접두어를 사용하는 멀티 밸류 옵션이에요. Map
bridgeErrorHandler (consumer) 컨슈머를 Camel 라우팅 에러 핸들러(Error Handler)에 브리지할 수 있게 해 줘요. 즉, Camel 컨슈머가 들어오는 메시지를 집어 올리는 동안 등에서 예외가 발생하면(가능한 경우), 그 예외를 메시지로 처리해서 라우팅 에러 핸들러가 다루게 돼요. 중요: 이것은 서드파티 컴포넌트가 예외가 던져졌을 때 Camel에 알릴 수 있도록 허용하는 경우에만 가능해요. 일부 컴포넌트는 내부에서만 처리하기 때문에 bridgeErrorHandler를 쓸 수 없어요. 다른 상황에서는 Camel 컴포넌트를 개선해서 서드파티 컴포넌트에 연결(hook)해 이를 가능하게 만들 수도 있어요. 기본으로 컨슈머는 org.apache.camel.spi.ExceptionHandler로 예외를 처리하는데, WARN 또는 ERROR 레벨로 로그를 남기고 무시해요. false boolean
configuration (consumer) 미리 구성된 설정(Configuration)을 지정할 수 있게 해 줘요. PostgresConnectorEmbeddedDebeziumConfiguration
internalKeyConverter (consumer) 오프셋의 키 데이터를 직렬화하고 역직렬화할 때 사용할 Converter 클래스예요. 기본값은 JSON 컨버터예요. org.apache.kafka.connect.json.JsonConverter String
internalValueConverter (consumer) 오프셋의 값 데이터를 직렬화하고 역직렬화할 때 사용할 Converter 클래스예요. 기본값은 JSON 컨버터예요. org.apache.kafka.connect.json.JsonConverter String
offsetCommitPolicy (consumer) 커밋 정책의 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
binaryHandlingMode (postgres) 변경 이벤트에서 binary(blob, binary 등) 컬럼을 어떻게 표현할지 정해요. 'bytes'는 binary 데이터를 byte 배열로 표현하고(기본값), 'base64'는 base64 인코딩 문자열로, 'base64-url-safe'는 base64-url-safe 인코딩 문자열로, 'hex'는 hex 인코딩(base16) 문자열로 표현해요. bytes String
columnExcludeList (postgres) 변경 이벤트에서 제외할 컬럼과 일치하는 정규 표현식이에요. String
columnIncludeList (postgres) 변경 이벤트에 포함할 컬럼과 일치하는 정규 표현식이에요. String
columnPropagateSourceType (postgres) 완전한 이름(fully-qualified name)으로 컬럼과 일치하는 정규 표현식을 쉼표로 나열한 목록이에요. 일치하면 해당 컬럼의 원래 타입과 원래 길이가 내보낸 변경 레코드의 필드 스키마에 파라미터로 추가돼요. String
connectionValidationTimeoutMs (postgres) 커넥션 검증이 끝나기까지 기다리는 최대 시간(밀리초)이에요. 기본값은 60초예요. 1m long
converters (postgres) 기본 컨버터 대신 쓸 커스텀 컨버터 목록(선택)이에요. 컨버터는 '.type' 설정 옵션으로 정의하고 '.' 옵션으로 설정해요. String
customMetricTags (postgres) 커스텀 메트릭 태그는 키-값 쌍을 받아서 MBean 객체 이름을 커스터마이즈해요. 태그는 일반 이름 끝에 붙고, 각 키는 MBean 객체 이름의 태그가 되고, 그에 대응하는 값이 그 키의 태그 값이 돼요. 예: k1=v1,k2=v2. String
customSanitizePattern (postgres) 값을 마스킹해야 할 설정 키를 식별하는 정규 표현식이에요. 이 값을 설정하면 Debezium의 기본 비밀번호 마스킹 패턴 대신 이 커스텀 패턴을 사용해요. .*secret$ .*password$
databaseDbname (postgres) 커넥터가 변경 사항을 캡처할 데이터베이스의 이름이에요. String
databaseHostname (postgres) 데이터베이스 서버의, 이름 확인이 가능한 호스트명이나 IP 주소예요. String
databaseInitialStatements (postgres) 데이터베이스에 JDBC 커넥션이 맺어졌을 때 실행할 SQL 문을 세미콜론으로 구분해서 나열한 목록이에요. 커넥터가 자체 판단으로 JDBC 커넥션을 맺을 수 있다는 점에 유의하세요. 그래서 보통은 DML 문 실행이 아니라 세션 파라미터 설정에만 쓰는 게 좋아요. 세미콜론을 구분자가 아니라 문자로 쓰려면 세미콜론을 두 개(';;') 쓰세요. String
databasePassword (postgres) 필수 데이터베이스에 접속할 때 쓸 데이터베이스 사용자의 비밀번호예요. String
databasePort (postgres) 데이터베이스 서버의 포트예요. 5432 int
databaseQueryTimeoutMs (postgres) 쿼리 실행을 기다리는 시간(밀리초)이에요. 기본값은 600초(600,000ms)이고, 0이면 제한이 없어요. 10m int
databaseSslcert (postgres) 클라이언트용 SSL 인증서가 담긴 파일이에요. 자세한 내용은 Postgres SSL 문서를 참고하세요. String
databaseSslfactory (postgres) SSL 소켓을 만드는 클래스의 이름이에요. 개발 환경에서 SSL 검증을 끄려면 org.postgresql.ssl.NonValidatingFactory를 사용하세요. String
databaseSslkey (postgres) 클라이언트용 SSL 개인 키가 담긴 파일이에요. 자세한 내용은 Postgres SSL 문서를 참고하세요. String
databaseSslmode (postgres) Postgres에 암호화 커넥션을 쓸지 여부예요. 옵션은 이렇게 있어요: 'disable'(기본값)은 암호화하지 않은 커넥션을 사용해요. 'allow'는 암호화하지 않은 커넥션을 먼저 시도하고, 안 되면 보안(암호화) 커넥션을 써요. 'prefer'(기본값)는 보안(암호화) 커넥션을 먼저 시도하고, 안 되면 암호화하지 않은 커넥션을 써요. 'require'는 보안(암호화) 커넥션을 쓰고, 맺을 수 없으면 실패해요. 'verify-ca'는 'required'와 비슷하지만 추가로 서버 TLS 인증서를 설정된 인증 기관(CA) 인증서와 대조해서 검증하고, 맞는 유효한 CA 인증서가 없으면 실패해요. 'verify-full'은 'verify-ca'와 비슷하지만 추가로 서버 인증서가 접속을 시도하는 호스트와 일치하는지 검증해요. prefer String
databaseSslpassword (postgres) 'database.sslkey'로 지정한 파일에서 클라이언트 개인 키에 접근할 때 쓸 비밀번호예요. 자세한 내용은 Postgres SSL 문서를 참고하세요. String
databaseSslrootcert (postgres) 서버 검증의 기준이 되는 루트 인증서(들)가 담긴 파일이에요. 자세한 내용은 Postgres JDBC SSL 문서를 참고하세요. String
databaseTcpkeepalive (postgres) TCP 커넥션이 끊기지 않도록 TCP keep-alive 프로브를 켤지 끌지 정해요. true boolean
databaseUser (postgres) 데이터베이스에 접속할 때 쓸 데이터베이스 사용자의 이름이에요. String
datatypePropagateSourceType (postgres) 데이터베이스 고유의 데이터 타입 이름과 일치하는 정규 표현식을 쉼표로 나열한 목록이에요. 일치하면 해당 데이터 타입의 원래 타입과 원래 길이가 내보낸 변경 레코드의 필드 스키마에 파라미터로 추가돼요. String
decimalHandlingMode (postgres) 변경 이벤트에서 DECIMAL과 NUMERIC 컬럼을 어떻게 표현할지 정해요. 'precise'(기본값)는 java.math.BigDecimal로 값을 표현하고, 변경 이벤트에는 binary 표현과 Kafka Connect의 'org.apache.kafka.connect.data.Decimal' 타입으로 인코딩돼요. 'string'은 값을 문자열로 표현해요. 'double'은 Java의 'double'로 표현하는데, 정밀도는 떨어질 수 있지만 컨슈머에서 훨씬 다루기 쉬워요. precise String
errorsMaxRetries (postgres) 실패하기 전에 커넥션 오류를 재시도하는 최대 횟수예요(-1 = 제한 없음, 0 = 비활성화, 0 = 재시도 횟수). -1 int
eventProcessingFailureHandlingMode (postgres) 이벤트 처리 중 실패(즉, 손상된 이벤트를 만났을 때)를 어떻게 다룰지 정해요. 'fail'(기본값)은 문제가 된 이벤트와 그 위치를 알리는 예외를 던져서 커넥터를 멈춰요. 'warn'은 문제가 된 이벤트와 그 위치를 로그로 남기고 그 이벤트를 건너뛰어요. 'ignore'는 문제가 된 이벤트를 그냥 건너뛰어요. fail String
executorShutdownTimeoutMs (postgres) 태스크 실행자(task executor)가 종료되기를 기다리는 최대 시간(밀리초)이에요. 4s long
extendedHeadersEnabled (postgres) 다운스트림 처리 시스템에서 CDC 이벤트의 출처를 추적·식별하는 데 꼭 필요한 메타데이터를 담는 Debezium 컨텍스트 헤더를 켜거나 꺼요. true boolean
guardrailCollectionsLimitAction (postgres) guardrail 컬렉션 한도를 초과했을 때 취할 동작을 정해요. 'warn'(기본값)은 경고 메시지를 로그로 남기고 계속 처리하고, 'fail'은 오류로 커넥터를 멈춰요. warn String
guardrailCollectionsMax (postgres) 커넥터가 캡처할 수 있는 컬렉션·테이블의 최대 개수예요. 한도를 초과하면 'guardrail.collections.limit.action'에 지정한 동작을 해요. 0으로 설정하면 이 guardrail을 끄는 거예요. 0 int
heartbeatActionQuery (postgres) 하트비트마다 실행하는 쿼리예요. String
heartbeatIntervalMs (postgres) 커넥터가 하트비트 토픽으로 하트비트 메시지를 주기적으로 보내는 간격(밀리초)이에요. 0으로 설정하면 하트비트 메시지를 꺼요. 기본값은 꺼져 있어요. 0ms int
heartbeatTopicsPrefix (postgres) 하트비트 토픽 이름에 쓰는 접두어예요. 기본값은 __debezium-heartbeat예요. __debezium-heartbeat String
hstoreHandlingMode (postgres) 변경 이벤트에서 HSTORE 컬럼을 어떻게 표현할지 정해요. 'json'은 값을 문자열화한 JSON으로 표현하고(기본값), 'map'은 키/값 맵으로 표현해요. json String
includeSchemaComments (postgres) 커넥터가 테이블·컬럼의 주석을 메타데이터 객체로 파싱할지 여부예요. 참고: 이 옵션을 켜면 메모리 사용량에 영향이 있어요. ColumnImpl 객체의 개수와 크기가 Debezium 커넥터의 메모리 소비를 크게 좌우하는데, 각 객체에 String을 하나 더하는 건 상당히 무거워질 수 있어요. 기본값은 'false'예요. false boolean
includeUnknownDatatypes (postgres) Debezium이 지원하지 않는 데이터 타입의 필드를 처리할지 정해요. 'false'(기본값)는 그 필드를 생략하고, 'true'는 구현에 따라 달라지는 binary 표현으로 필드를 변환해요. false boolean
incrementalSnapshotChunkSize (postgres) 증분 스냅샷에서 청크(chunk)의 최대 크기(문서/로우 수)예요. 1024 int
incrementalSnapshotWatermarkingStrategy (postgres) 증분 스냅샷 중 워터마킹에 쓰는 전략을 정해요. 'insert_insert'는 open 신호와 close 신호 둘 다 시그널 데이터 컬렉션에 기록하고(기본값), 'insert_delete'는 open 신호만 기록하고 close가 해당 open 신호를 지워요. INSERT_INSERT String
intervalHandlingMode (postgres) 변경 이벤트에서 INTERVAL 컬럼을 어떻게 표현할지 정해요. 'string'은 정확한 ISO 형식 문자열로 표현하고, 'numeric'(기본값)은 정확하지 않은 마이크로초 변환으로 표현해요. numeric String
legacySnapshotMaxThreads (postgres) 켜면 레거시 table-per-thread 병렬 스냅샷 알고리즘을 사용해요. false로 두면(기본값) 테이블을 청크로 나눠서 모든 스냅샷 스레드에 걸쳐 처리하기 때문에 스냅샷 동시성을 더 높일 수 있어요. false boolean
lsnFlushMode (postgres) LSN 플러시 전략을 정해요. 옵션은 이렇게 있어요: 'connector'(기본값)는 Debezium이 관리하는 LSN 플러시예요(폐기 예정인 flush.lsn.source=true를 대체). 'manual'은 외부에서 관리하는 LSN 플러시예요(폐기 예정인 flush.lsn.source=false를 대체). 'connector_and_driver'는 Debezium이 관리하는 LSN 플러시에 pgjdbc 드라이버가 서버 keepalive LSN으로 감시하지 않는(unmonitored) LSN까지 플러시하는 방식이고, 활동이 적은 데이터베이스에서 WAL이 자라는 걸 막아 줘요. String
lsnFlushTimeoutAction (postgres) LSN 플러시 타임아웃이 발생했을 때 취할 동작이에요. 옵션은 이렇게 있어요: 'fail'(기본값)은 커넥터를 실패시키고, 'warn'은 경고를 로그로 남기고 계속 처리하며, 'ignore'는 타임아웃을 무시하고 계속 처리해요. fail String
lsnFlushTimeoutMs (postgres) LSN 플러시 동작이 끝나기를 기다리는 최대 시간(밀리초)이에요. 이 시간 안에 플러시가 끝나지 않으면 lsn.flush.timeout.action에 지정한 동작을 해요. 기본값은 30초예요. 30s long
maxBatchSize (postgres) 소스 레코드 배치 하나의 최대 크기예요. 기본값은 2048이에요. 2048 int
maxQueueSize (postgres) 데이터베이스 로그에서 읽었지만 아직 기록되거나 전달되지 않은 변경 이벤트 큐의 최대 크기예요. 기본값은 8192이고, 항상 최대 배치 크기보다 커야 해요. 8192 int
maxQueueSizeInBytes (postgres) 같은 큐의 바이트 단위 최대 크기예요. 기본값은 0이고, 이는 이 기능이 꺼져 있다는 뜻이에요. 0 long
memoryManagementSchemasClass (postgres) 스키마 메타데이터용 스토리지 구현의 완전한 클래스 이름(fully-qualified class name)이에요. 이 클래스는 io.debezium.relational.TableMappingStorage를 구현해야 하고, 기본값은 인메모리 스토리지용 io.debezium.relational.ConcurrentMapTableMappingStorage예요. io.debezium.relational.ConcurrentMapTableMappingStorage String
memoryManagementTablesClass (postgres) 테이블 메타데이터용 스토리지 구현의 완전한 클래스 이름(fully-qualified class name)이에요. 이 클래스는 io.debezium.relational.TableMappingStorage를 구현해야 하고, 기본값은 인메모리 스토리지용 io.debezium.relational.ConcurrentMapTableMappingStorage예요. io.debezium.relational.ConcurrentMapTableMappingStorage String
messageKeyColumns (postgres) 메시지 키로 쓸 완전한 이름(fully-qualified)의 테이블과 컬럼(들)과 일치하는 표현식을 세미콜론으로 나열한 목록이에요. 각 표현식은 ':' 패턴과 일치해야 하고, 테이블 이름은 커넥터에 따라 (DB_NAME.TABLE_NAME)이나 (SCHEMA_NAME.TABLE_NAME)으로 정의할 수 있으며, 키 컬럼은 커스텀 키를 나타내는 컬럼의 쉼표 목록이에요. 명시적인 키 설정이 없는 테이블은 테이블의 기본 키 컬럼(들)이 메시지 키로 쓰여요. 예: dbserver1.inventory.orderlines:orderId,orderLineId;dbserver1.inventory.orders:id. String
messagePrefixExcludeList (postgres) 모니터링에서 제외할 논리 디코딩(logical decoding) 메시지 접두어와 일치하는 정규 표현식을 쉼표로 나열한 목록이에요. String
messagePrefixIncludeList (postgres) 모니터링할 논리 디코딩(logical decoding) 메시지 접두어와 일치하는 정규 표현식을 쉼표로 나열한 목록이에요. 기본값으로 모든 접두어를 모니터링해요. String
notificationEnabledChannels (postgres) 활성화할 알림(notifcation) 채널 이름의 목록이에요. String
notificationSinkTopicName (postgres) 알림을 담을 토픽의 이름이에요. 활성화된 채널 목록에 'sink'가 있으면 필수예요. String
openlineageIntegrationConfigFilePath (postgres) OpenLineage 파일 설정의 경로예요. 자세한 내용은 https://openlineage.io/docs/client/java/configuration을 참고하세요. ./openlineage.yml String
openlineageIntegrationDatasetKafkaBootstrapServers (postgres) 입력/출력 네임스페이스로 쓰이는 Kafka 부트스트랩 서버 주소예요. String
openlineageIntegrationEnabled (postgres) Debezium이 OpenLineage API로 데이터 리니지(data lineage) 메타데이터를 내보내도록 켜요. false boolean
openlineageIntegrationJobDescription (postgres) Debezium이 내보내는 job의 설명이에요. Debezium change data capture job String
openlineageIntegrationJobNamespace (postgres) Debezium이 내보내는 job의 네임스페이스예요. String
openlineageIntegrationJobOwners (postgres) Debezium이 내보내는 job의 소유자(owners)예요. 키-값 쌍을 쉼표로 나열해요. 예: k1=v1,k2=v2. String
openlineageIntegrationJobTags (postgres) Debezium이 내보내는 job의 태그예요. 키-값 쌍을 쉼표로 나열해요. 예: k1=v1,k2=v2. String
pluginName (postgres) 서버에 설치된 Postgres 논리 디코딩(logical decoding) 플러그인의 이름이에요. 지원하는 값은 'decoderbufs'와 'pgoutput'이고, 기본값은 'decoderbufs'예요. decoderbufs String
pollIntervalMs (postgres) 이벤트가 없을 때 새 변경 이벤트가 나타나기를 기다리는 시간(밀리초)이에요. 기본값은 500ms예요. 500ms long
postProcessors (postgres) 포스트 프로세서(post processor) 목록(선택)이에요. 프로세서는 '.type' 설정 옵션으로 정의하고 '' 옵션으로 설정해요. String
provideTransactionMetadata (postgres) 이벤트 카운트와 함께 트랜잭션 메타데이터 추출을 켜요. false boolean
publicationAutocreateMode (postgres) pgoutput으로 변경을 스트리밍할 때만 적용돼요. publication 생성을 어떻게 동작하게 할지 정하고, 기본값은 all_tables예요. DISABLED - 커넥터가 publication을 전혀 만들려고 하지 않아요. 사용자가 미리 publication을 만들어 뒀다고 가정해요. 시작할 때 publication이 없으면 커넥터는 예외를 던지고 멈춰요. ALL_TABLES - publication이 없으면 모든 테이블에 대한 새 publication을 만들어요. 이때 설정된 사용자가 접근 권한이 있어야 한다는 점에 유의하세요. publication이 이미 있으면 그걸 사용해요. 즉 CREATE PUBLICATION FOR ALL TABLES예요. FILTERED - publication이 없으면 현재 필터 설정(테이블/데이터베이스 include/exclude 목록 프로퍼티 참고)에 맞는 테이블들에 대한 새 publication을 만들어요. publication이 이미 있으면 그걸 사용해요. 즉 CREATE PUBLICATION FOR TABLE이에요. all_tables String
publicationName (postgres) 플러그인에서 변경을 스트리밍할 때 쓰는 Postgres 10 publication의 이름이에요. 기본값은 'dbz_publication'예요. dbz_publication String
publishViaPartitionRoot (postgres) 커넥터가 파티션 루트(partition root)를 통해 변경을 발행할지 정하는 불리언이에요. true면 파티션 루트를 통해 발행하고, false면 바로 발행해요. false boolean
queryFetchSize (postgres) 스트리밍 중에 메모리로 읽어 들일 최대 레코드 수예요. '0'이면 기본 JDBC fetch size를 사용해요. 0 int
replicaIdentityAutosetValues (postgres) pgoutput으로 변경을 스트리밍할 때만 적용돼요. 테이블 수준의 Replica Identity 값을 정해요. 이 옵션은 데이터베이스에 이미 있는 값을 덮어써요. 완전한 이름(fully-qualified)의 테이블과 그 테이블에 사용할 Replica Identity 값이 일치하는 정규 표현식을 쉼표로 나열한 목록이에요. 각 표현식은 ':' 패턴과 일치해야 하고, 테이블 이름은 (SCHEMA_NAME.TABLE_NAME)으로 정의해요. Replica Identity 값은 이렇게 있어요: DEFAULT - 기본 키 컬럼의 이전 값을 기록해요(있다면). 시스템이 아닌 테이블의 기본값이에요. INDEX index_name - 지정한 인덱스가 다루는 컬럼의 이전 값을 기록해요. 이 인덱스는 유니크해야 하고, partial이 아니어야 하고, deferrable이 아니어야 하며, NOT NULL로 표시된 컬럼만 포함해야 해요. 이 인덱스를 지우면 NOTHING과 같은 동작을 해요. FULL - 로우의 모든 컬럼에 대한 이전 값을 기록해요. NOTHING - 이전 로우에 대한 정보를 기록하지 않아요. 시스템 테이블의 기본값이에요. String
retriableRestartConnectorWaitMs (postgres) 재시도 가능한(retriable) 예외가 발생한 뒤 커넥터를 다시 시작하기 전에 기다리는 시간이에요. 기본값은 10000ms예요. 10s long
schemaExcludeList (postgres) 이벤트를 캡처하면 안 되는 스키마들이에요. String
schemaHistoryInternalFileFilename (postgres) 데이터베이스 스키마 히스토리를 기록하는 데 쓸 파일의 경로예요. String
schemaIncludeList (postgres) 이벤트를 캡처할 스키마들이에요. String
schemaNameAdjustmentMode (postgres) 커넥터가 사용하는 메시지 컨버터와의 호환성을 위해 스키마 이름을 어떻게 조정할지 정해요. 'avro'는 Avro 타입 이름에 쓸 수 없는 문자를 언더스코어로 바꿔요. 'avro_unicode'는 언더스코어나 Avro 타입 이름에 쓸 수 없는 문자를 _uxxxx 같은 대응하는 유니코드로 바꿔요. 참고: _는 Java의 백슬래시 같은 이스케이프 시퀀스예요. 'none'은 아무 조정도 하지 않아요(기본값). none String
schemaRefreshMode (postgres) 테이블의 인메모리 스키마를 새로 고치는 조건을 정해요. 'columns_diff'(기본값)가 가장 안전한 모드로, 인메모리 스키마가 항상 데이터베이스 테이블 스키마와 동기화되도록 보장해요. 'columns_diff_exclude_unchanged_toast'는 인메모리 스키마 캐시와 들어오는 메시지에서 도출한 스키마 사이에 불일치가 있을 때 캐시를 새로 고치라고 지시하지만, 변경되지 않은 TOASTable 데이터가 그 불일치를 전부 설명한다면 예외예요. 자주 업데이트되는 테이블에 TOASTed 데이터가 있고 그 데이터가 이런 업데이트에 잘 끼어들지 않는다면 이 설정은 커넥터 성능을 크게 개선할 수 있어요. 다만 TOASTable 컬럼을 테이블에서 지우면 인메모리 스키마가 낡을 수 있어요. columns_diff String
signalDataCollection (postgres) Debezium에 시그널/커맨드를 보낼 때 쓰는 데이터 컬렉션의 이름이에요. 멀티 파티션 모드 커넥터는 시그널 데이터 컬렉션을 쉼표로 구분해서 여러 개 지정할 수 있어요. 설정하지 않으면 시그널링은 꺼져 있어요. String
signalEnabledChannels (postgres) 활성화된 채널 이름의 목록이에요. source 채널이 기본으로 활성화돼요. source String
signalPollIntervalMs (postgres) 등록된 채널에서 새 시그널을 찾는 간격(밀리초)이에요. 기본값은 5초예요. 5s long
skippedOperations (postgres) 스트리밍 중에 건너뛸 연산을 쉼표로 나열한 목록이에요. 'c'는 insert/create, 'u'는 update, 'd'는 delete, 't'는 truncate를 뜻하고, 'none'은 아무것도 건너뛰지 않는다는 뜻이에요. 기본값으로는 truncate 연산만 건너뛰어요. t String
slotDropOnStop (postgres) 커넥터가 정상적으로 끝났을 때 논리 복제 슬롯(logical replication slot)을 지울지 여부예요. 기본값으로는 복제 슬롯을 유지해서, 재시작할 때 마지막으로 기록된 위치부터 이어서 진행할 수 있게 해요. false boolean
slotFailover (postgres) 페일오버 슬롯(failover slot)을 만들지 여부예요. Postgres 클러스터의 프라이머리 서버에 접속할 때, 버전 17 이상에서만 지원해요. 지정하지 않거나 Postgres 17 프라이머리에 접속하지 않으면 페일오버 슬롯을 만들지 않아요. false boolean
slotMaxRetries (postgres) 복제 슬롯 접속 시도가 실패했을 때 재시도하는 횟수예요. 6 int
slotName (postgres) 플러그인에서 변경을 스트리밍하려고 만든 Postgres 논리 디코딩(logical decoding) 슬롯의 이름이에요. 기본값은 'debezium이에요. debezium String
slotRetryDelayMs (postgres) 커넥터가 복제 슬롯 접속에 실패했을 때 재시도 사이에 기다리는 시간(밀리초)이에요. 기본값은 10초(10,000ms)예요. 10s long
slotStreamParams (postgres) 논리 디코딩(logical decoding) 플러그인이 쓰는 선택적 파라미터예요. 세미콜론으로 구분해요. 예: 'add-tables=public.table,public.table2;include-lsn=true'. String
snapshotDelayMs (postgres) 스냅샷이 시작되기 전의 지연 시간(밀리초)이에요. 기본값은 0ms예요. 0ms long
snapshotFetchSize (postgres) 스냅샷을 수행하는 동안 메모리로 읽어 들일 최대 레코드 수예요. int
snapshotIncludeCollectionList (postgres) 커넥터를 만들거나 재시작할 때 스냅샷을 찍어야 할 테이블/컬렉션 목록을 지정하려면 이 설정을 반드시 지정해야 해요. String
snapshotIsolationMode (postgres) 어떤 트랜잭션 격리 수준을 쓸지 조절해요. 기본값은 'serializable'로, serializable 격리 수준을 쓴다는 뜻이에요. 'repeatable_read'를 지정하면 커넥터가 초기 스냅샷을 REPEATABLE READ 격리 수준으로 수행해요. 'read_committed'를 지정하면 READ COMMITTED 격리 수준으로 수행해요. 'read_uncommitted'를 지정하면 READ UNCOMMITTED 격리 수준으로 수행해요. serializable String
snapshotLockingMode (postgres) 스키마 스냅샷을 수행하는 동안 커넥터가 테이블 잠금을 어떻게 쥐고 있을지 조절해요. 'shared'는 커넥터가 데이터베이스 스키마와 다른 메타데이터를 읽는 스냅샷 초기 구간에만 배타적 테이블 접근을 막는 테이블 잠금을 쥐고 있게 해요. 스냅샷의 나머지 작업은 각 테이블에서 모든 로우를 select하는 건데, 이때는 잠금이 필요 없는 flashback query를 사용해요. 하지만 어떤 경우엔 잠금을 아예 피하는 게 좋을 수도 있는데, 'none'을 지정하면 그렇게 할 수 있어요. 이 모드는 스냅샷을 찍는 동안 스키마 변경이 일어나지 않을 때만 안전해요. none String
snapshotLockingModeCustomName (postgres) 'snapshot.locking.mode'가 custom으로 설정되어 있으면, 'name()' 메서드가 제공하는 커스텀 구현의 이름을 지정하도록 이 설정을 반드시 지정해야 해요. 구현체는 'SnapshotterLocking' 인터페이스를 구현해야 하고, 스키마 스냅샷 중 테이블을 어떻게 잠글지 정하려고 호출돼요. String
snapshotLockTimeoutMs (postgres) 스냅샷 시작 시 테이블 잠금을 기다리는 최대 밀리초예요. 이 시간 안에 잠금을 얻지 못하면 스냅샷이 중단돼요. 기본값은 10초예요. 10s long
snapshotMaxThreads (postgres) 스냅샷을 수행할 때 쓰는 최대 스레드 수예요. 기본값은 1이에요. 1 int
snapshotMaxThreadsMultiplier (postgres) 테이블당 스냅샷 청크 개수를 늘리는 데 쓰는 계수예요. 기본 동작은 'row_count/snapshot.max.threads'로 로우당 청크 수를 계산해요. 큰 테이블에는 이게 최선이 아닐 수 있는데, multiplier를 쓰면 'row_count/(snapshot.max.threads snapshot.max.threads.multiplier)'로 청크 개수를 늘리도록 공식이 조정돼요. 1 int
snapshotMode (postgres) 커넥터 시작 시 스냅샷을 실행하는 기준이에요. 다음 스냅샷 옵션 중 하나를 고르세요: 'always': 커넥터가 시작할 때마다 스냅샷을 실행해요. 스냅샷이 끝나면 트랜잭션 로그에서 변경을 스트리밍하기 시작해요. 'initial'(기본값): 커넥터가 논리 서버 이름에 대한 오프셋을 찾지 못하면, 설정된 테이블의 현재 전체 상태를 담는 스냅샷을 실행해요. 스냅샷이 끝나면 트랜잭션 로그에서 변경을 스트리밍하기 시작해요. 'initial_only': 'initial' 옵션처럼 스냅샷을 수행하지만, 스냅샷이 끝나면 커넥터가 멈추고 트랜잭션 로그에서 변경을 스트리밍하지 않아요. 'never': 커넥터가 스냅샷을 실행하지 않아요. 처음 시작하면 즉시 트랜잭션 로그의 처음부터 읽기 시작해요. 'exported': 이 옵션은 폐기 예정이에요. 대신 'initial'을 사용하세요. 'custom': 커넥터가 스냅샷을 수행하는 방법을 지정할 커스텀 클래스를 로드해요. 자세한 내용은 PostgreSQL 커넥터 문서의 Custom snapshotter SPI를 참고하세요. initial String
snapshotModeConfigurationBasedSnapshotData (postgres) 'snapshot.mode'가 configuration_based로 설정되어 있으면, 데이터를 스냅샷할지 말지를 지정할 수 있게 해 줘요. false boolean
snapshotModeConfigurationBasedSnapshotOnDataError (postgres) 'snapshot.mode'가 configuration_based로 설정되어 있으면, 오류가 났을 때 데이터를 스냅샷할지 말지를 지정할 수 있게 해 줘요. false boolean
snapshotModeConfigurationBasedSnapshotOnSchemaError (postgres) 'snapshot.mode'가 configuration_based로 설정되어 있으면, 오류가 났을 때 스키마를 스냅샷할지 말지를 지정할 수 있게 해 줘요. false boolean
snapshotModeConfigurationBasedSnapshotSchema (postgres) 'snapshot.mode'가 configuration_based로 설정되어 있으면, 스키마를 스냅샷할지 말지를 지정할 수 있게 해 줘요. false boolean
snapshotModeConfigurationBasedStartStream (postgres) 'snapshot.mode'가 configuration_based로 설정되어 있으면, 스냅샷 이후에 스트림을 시작할지 말지를 지정할 수 있게 해 줘요. false boolean
snapshotModeCustomName (postgres) 'snapshot.mode'가 custom으로 설정되어 있으면, 'name()' 메서드가 제공하는 커스텀 구현의 이름을 지정하도록 이 설정을 반드시 지정해야 해요. 구현체는 'Snapshotter' 인터페이스를 구현해야 하고, 앱이 부팅될 때마다 호출돼서 스냅샷을 할지 정해요. String
snapshotQueryMode (postgres) 스냅샷 동안 사용할 쿼리를 조절해요. select_all String
snapshotQueryModeCustomName (postgres) 'snapshot.query.mode'가 custom으로 설정되어 있으면, 'name()' 메서드가 제공하는 커스텀 구현의 이름을 지정하도록 이 설정을 반드시 지정해야 해요. 구현체는 'SnapshotterQuery' 인터페이스를 구현해야 하고, 스냅샷 동안 쿼리를 어떻게 만들지 정하려고 호출돼요. String
snapshotSelectStatementOverrides (postgres) 이 프로퍼티는 완전한 이름(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 (postgres) 초기 스냅샷에서 테이블을 처리하는 순서를 조절해요. descending 값을 주면 테이블을 로우 수 내림차순으로 정렬해요. ascending 값을 주면 로우 수 오름차순으로 정렬해요. disabled(기본값)는 로우 수 정렬을 꺼요. disabled String
sourceinfoStructMaker (postgres) SourceInfo 스키마와 struct를 돌려주는 SourceInfoStructMaker 클래스의 이름이에요. io.debezium.connector.postgresql.PostgresSourceInfoStructMaker String
statisticsMetricsEnabled (postgres) 레코드 처리 지연 시간 같은 여러 통계와 quantiles 같은 파생 데이터를 수집하게 해요. 기본값으로 통계 수집은 켜져 있어요. true boolean
statusUpdateIntervalMs (postgres) 복제 커넥션 상태 업데이트를 서버로 보내는 빈도(밀리초)예요. 기본값은 10초(10,000ms)예요. 10s int
streamingDelayMs (postgres) 스냅샷이 끝나고 스트리밍이 시작되기까지의 지연 시간(밀리초)이에요. 기본값은 0ms예요. 0ms long
tableExcludeList (postgres) 모니터링에서 제외할 테이블의 완전한 이름(fully-qualified name)과 일치하는 정규 표현식을 쉼표로 나열한 목록이에요. String
tableIgnoreBuiltin (postgres) 내장(built-in) 테이블을 무시할지 여부를 지정하는 플래그예요. true boolean
tableIncludeList (postgres) 변경을 캡처할 테이블들이에요. String
timePrecisionMode (postgres) 시간, 날짜, 타임스탬프는 여러 정밀도로 표현할 수 있어요. 'adaptive'(기본값)는 시간·날짜·타임스탬프 값의 정밀도를 데이터베이스 컬럼의 정밀도에 맞춰요. 'adaptive_time_microseconds'는 'adaptive' 모드와 비슷하지만 TIME 필드는 항상 마이크로초 정밀도를 써요. 'connect'는 데이터베이스 컬럼의 정밀도와 무관하게 항상 밀리초 정밀도의 Kafka Connect 내장 Time·Date·Timestamp 표현으로 표현해요. 'isostring'은 시간·날짜·타임스탬프 값을 UTC 시간대의 ISO-8601 형식 문자열로 표현해요. 'microseconds'는 항상 마이크로초 정밀도로 표현해요. 'nanoseconds'는 항상 나노초 정밀도로 표현해요. adaptive String
tombstonesOnDelete (postgres) delete 연산을 delete 이벤트와 그 뒤의 tombstone 이벤트로 표현할지(true), delete 이벤트만으로 표현할지(false) 여부예요. tombstone 이벤트를 내보내면(기본 동작) 소스 레코드가 지워졌을 때 Kafka가 해당 키와 관련된 모든 이벤트를 완전히 삭제할 수 있어요. false boolean
topicNamingStrategy (postgres) 데이터 변경, 스키마 변경, 트랜잭션, 하트비트 이벤트 등의 토픽 이름을 정하는 데 쓸 TopicNamingStrategy 클래스의 이름이에요. io.debezium.schema.SchemaTopicNamingStrategy String
topicPrefix (postgres) 필수 특정 데이터베이스 서버/클러스터가 변경을 캡처하고 있음을 식별하고 네임스페이스를 제공하는 토픽 접두어예요. 토픽 접두어는 다른 모든 커넥터와 구별되게 유니크해야 해요. 이 커넥터가 내보내는 이벤트를 받는 모든 Kafka 토픽 이름의 접두어로 쓰이기 때문이에요. 영숫자, 하이픈, 점, 언더스코어만 받아들여야 해요. String
transactionMetadataFactory (postgres) 트랜잭션 컨텍스트와 트랜잭션 struct/스키마를 만드는 클래스예요. io.debezium.pipeline.txmetadata.DefaultTransactionMetadataFactory String
unavailableValuePlaceholder (postgres) 원래 값이 데이터베이스가 제공하지 않는 toasted 값임을 나타내는 상수를 정해요. 'hex:' 접두어로 시작하면 나머지 문자열이 16진수로 인코딩된 옥텟을 뜻하는 걸로 기대해요. __debezium_unavailable_value String
xminFetchIntervalMs (postgres) 복제 슬롯에서 xmin을 얼마나 자주(밀리초) 가져올지 정해요. 이 xmin 값은 슬롯이 노출하는 값으로, 새 복제 슬롯이 어디서 시작할 수 있는지의 하한(lower bound)이 돼요. 값이 작을수록 이 값이 현재의 '참' 값일 가능성이 높지만 성능 비용은 커져요. 값이 클수록 현재의 '참' 값일 가능성은 낮아지지만 성능 부담은 줄어들어요. 기본값은 0ms로, xmin 추적을 꺼요. 0ms long

엔드포인트 옵션 (Endpoint Options)

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

debezium-postgres:name

다음 _path_와 query 파라미터를 사용해요:

경로 파라미터 (Path Parameters)

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

쿼리 파라미터 (Query Parameters)

Name Description Default Type
additionalProperties (common) camel 설정에 바로 넣을 수 없는 debezium 컴포넌트용 추가 프로퍼티예요(예: Debezium 엔진에 필요한 Kafka Connect 프로퍼티 설정, KafkaOffsetBackingStore 설정 같은 것). 프로퍼티에는 additionalProperties. 접두어를 붙여야 해요. 예: additionalProperties.transactional.id=12345&additionalProperties.schema.registry.url=http://localhost:8811/avro. additionalProperties. 접두어를 사용하는 멀티 밸류 옵션이에요. Map
internalKeyConverter (consumer) 오프셋의 키 데이터를 직렬화하고 역직렬화할 때 사용할 Converter 클래스예요. 기본값은 JSON 컨버터예요. org.apache.kafka.connect.json.JsonConverter String
internalValueConverter (consumer) 오프셋의 값 데이터를 직렬화하고 역직렬화할 때 사용할 Converter 클래스예요. 기본값은 JSON 컨버터예요. org.apache.kafka.connect.json.JsonConverter String
offsetCommitPolicy (consumer) 커밋 정책의 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 | | binaryHandlingMode (postgres) | 변경 이벤트에서 binary(blob, binary 등) 컬럼을 어떻게 표현할지 정해요. 'bytes'는 binary 데이터를 byte 배열로 표현하고(기본값), 'base64'는 base64 인코딩 문자열로, 'base64-url-safe'는 base64-url-safe 인코딩 문자열로, 'hex'는 hex 인코딩(base16) 문자열로 표현해요. | bytes | String | | columnExcludeList (postgres) | 변경 이벤트에서 제외할 컬럼과 일치하는 정규 표현식이에요. | | String | | columnIncludeList (postgres) | 변경 이벤트에 포함할 컬럼과 일치하는 정규 표현식이에요. | | String | | columnPropagateSourceType (postgres) | 완전한 이름(fully-qualified name)으로 컬럼과 일치하는 정규 표현식을 쉼표로 나열한 목록이에요. 일치하면 해당 컬럼의 원래 타입과 원래 길이가 내보낸 변경 레코드의 필드 스키마에 파라미터로 추가돼요. | | String | | connectionValidationTimeoutMs (postgres) | 커넥션 검증이 끝나기까지 기다리는 최대 시간(밀리초)이에요. 기본값은 60초예요. | 1m | long | | converters (postgres) | 기본 컨버터 대신 쓸 커스텀 컨버터 목록(선택)이에요. 컨버터는 '.type' 설정 옵션으로 정의하고 '.' 옵션으로 설정해요. | | String | | customMetricTags (postgres) | 커스텀 메트릭 태그는 키-값 쌍을 받아서 MBean 객체 이름을 커스터마이즈해요. 태그는 일반 이름 끝에 붙고, 각 키는 MBean 객체 이름의 태그가 되고, 그에 대응하는 값이 그 키의 태그 값이 돼요. 예: k1=v1,k2=v2. | | String | | customSanitizePattern (postgres) | 값을 마스킹해야 할 설정 키를 식별하는 정규 표현식이에요. 이 값을 설정하면 Debezium의 기본 비밀번호 마스킹 패턴 대신 이 커스텀 패턴을 사용해요. | .*secret$|.*password$|.*sasl\.jaas\.config$|.*basic\.auth\.user\.info|.*registry\.auth\.client-secret|.*credentials\.json$|.*connectionstring$|.*connection\.string$ | String | | databaseDbname (postgres) | 커넥터가 변경 사항을 캡처할 데이터베이스의 이름이에요. | | String | | databaseHostname (postgres) | 데이터베이스 서버의, 이름 확인이 가능한 호스트명이나 IP 주소예요. | | String | | databaseInitialStatements (postgres) | 데이터베이스에 JDBC 커넥션이 맺어졌을 때 실행할 SQL 문을 세미콜론으로 구분해서 나열한 목록이에요. 커넥터가 자체 판단으로 JDBC 커넥션을 맺을 수 있다는 점에 유의하세요. 그래서 보통은 DML 문 실행이 아니라 세션 파라미터 설정에만 쓰는 게 좋아요. 세미콜론을 구분자가 아니라 문자로 쓰려면 세미콜론을 두 개(';;') 쓰세요. | | String | | databasePassword (postgres) | 필수 데이터베이스에 접속할 때 쓸 데이터베이스 사용자의 비밀번호예요. | | String | | databasePort (postgres) | 데이터베이스 서버의 포트예요. | 5432 | int | | databaseQueryTimeoutMs (postgres) | 쿼리 실행을 기다리는 시간(밀리초)이에요. 기본값은 600초(600,000ms)이고, 0이면 제한이 없어요. | 10m | int | | databaseSslcert (postgres) | 클라이언트용 SSL 인증서가 담긴 파일이에요. 자세한 내용은 Postgres SSL 문서를 참고하세요. | | String | | databaseSslfactory (postgres) | SSL 소켓을 만드는 클래스의 이름이에요. 개발 환경에서 SSL 검증을 끄려면 org.postgresql.ssl.NonValidatingFactory를 사용하세요. | | String | | databaseSslkey (postgres) | 클라이언트용 SSL 개인 키가 담긴 파일이에요. 자세한 내용은 Postgres SSL 문서를 참고하세요. | | String | | databaseSslmode (postgres) | Postgres에 암호화 커넥션을 쓸지 여부예요. 옵션은 이렇게 있어요: 'disable'(기본값)은 암호화하지 않은 커넥션을 사용해요. 'allow'는 암호화하지 않은 커넥션을 먼저 시도하고, 안 되면 보안(암호화) 커넥션을 써요. 'prefer'(기본값)는 보안(암호화) 커넥션을 먼저 시도하고, 안 되면 암호화하지 않은 커넥션을 써요. 'require'는 보안(암호화) 커넥션을 쓰고, 맺을 수 없으면 실패해요. 'verify-ca'는 'required'와 비슷하지만 추가로 서버 TLS 인증서를 설정된 인증 기관(CA) 인증서와 대조해서 검증하고, 맞는 유효한 CA 인증서가 없으면 실패해요. 'verify-full'은 'verify-ca'와 비슷하지만 추가로 서버 인증서가 접속을 시도하는 호스트와 일치하는지 검증해요. | prefer | String | | databaseSslpassword (postgres) | 'database.sslkey'로 지정한 파일에서 클라이언트 개인 키에 접근할 때 쓸 비밀번호예요. 자세한 내용은 Postgres SSL 문서를 참고하세요. | | String | | databaseSslrootcert (postgres) | 서버 검증의 기준이 되는 루트 인증서(들)가 담긴 파일이에요. 자세한 내용은 Postgres JDBC SSL 문서를 참고하세요. | | String | | databaseTcpkeepalive (postgres) | TCP 커넥션이 끊기지 않도록 TCP keep-alive 프로브를 켤지 끌지 정해요. | true | boolean | | databaseUser (postgres) | 데이터베이스에 접속할 때 쓸 데이터베이스 사용자의 이름이에요. | | String | | datatypePropagateSourceType (postgres) | 데이터베이스 고유의 데이터 타입 이름과 일치하는 정규 표현식을 쉼표로 나열한 목록이에요. 일치하면 해당 데이터 타입의 원래 타입과 원래 길이가 내보낸 변경 레코드의 필드 스키마에 파라미터로 추가돼요. | | String | | decimalHandlingMode (postgres) | 변경 이벤트에서 DECIMAL과 NUMERIC 컬럼을 어떻게 표현할지 정해요. 'precise'(기본값)는 java.math.BigDecimal로 값을 표현하고, 변경 이벤트에는 binary 표현과 Kafka Connect의 'org.apache.kafka.connect.data.Decimal' 타입으로 인코딩돼요. 'string'은 값을 문자열로 표현해요. 'double'은 Java의 'double'로 표현하는데, 정밀도는 떨어질 수 있지만 컨슈머에서 훨씬 다루기 쉬워요. | precise | String | | errorsMaxRetries (postgres) | 실패하기 전에 커넥션 오류를 재시도하는 최대 횟수예요(-1 = 제한 없음, 0 = 비활성화, 0 = 재시도 횟수). | -1 | int | | eventProcessingFailureHandlingMode (postgres) | 이벤트 처리 중 실패(즉, 손상된 이벤트를 만났을 때)를 어떻게 다룰지 정해요. 'fail'(기본값)은 문제가 된 이벤트와 그 위치를 알리는 예외를 던져서 커넥터를 멈춰요. 'warn'은 문제가 된 이벤트와 그 위치를 로그로 남기고 그 이벤트를 건너뛰어요. 'ignore'는 문제가 된 이벤트를 그냥 건너뛰어요. | fail | String | | executorShutdownTimeoutMs (postgres) | 태스크 실행자(task executor)가 종료되기를 기다리는 최대 시간(밀리초)이에요. | 4s | long | | extendedHeadersEnabled (postgres) | 다운스트림 처리 시스템에서 CDC 이벤트의 출처를 추적·식별하는 데 꼭 필요한 메타데이터를 담는 Debezium 컨텍스트 헤더를 켜거나 꺼요. | true | boolean | | guardrailCollectionsLimitAction (postgres) | guardrail 컬렉션 한도를 초과했을 때 취할 동작을 정해요. 'warn'(기본값)은 경고 메시지를 로그로 남기고 계속 처리하고, 'fail'은 오류로 커넥터를 멈춰요. | warn | String | | guardrailCollectionsMax (postgres) | 커넥터가 캡처할 수 있는 컬렉션·테이블의 최대 개수예요. 한도를 초과하면 'guardrail.collections.limit.action'에 지정한 동작을 해요. 0으로 설정하면 이 guardrail을 끄는 거예요. | 0 | int | | heartbeatActionQuery (postgres) | 하트비트마다 실행하는 쿼리예요. | | String | | heartbeatIntervalMs (postgres) | 커넥터가 하트비트 토픽으로 하트비트 메시지를 주기적으로 보내는 간격(밀리초)이에요. 0으로 설정하면 하트비트 메시지를 꺼요. 기본값은 꺼져 있어요. | 0ms | int | | heartbeatTopicsPrefix (postgres) | 하트비트 토픽 이름에 쓰는 접두어예요. 기본값은 __debezium-heartbeat예요. | __debezium-heartbeat | String | | hstoreHandlingMode (postgres) | 변경 이벤트에서 HSTORE 컬럼을 어떻게 표현할지 정해요. 'json'은 값을 문자열화한 JSON으로 표현하고(기본값), 'map'은 키/값 맵으로 표현해요. | json | String | | includeSchemaComments (postgres) | 커넥터가 테이블·컬럼의 주석을 메타데이터 객체로 파싱할지 여부예요. 참고: 이 옵션을 켜면 메모리 사용량에 영향이 있어요. ColumnImpl 객체의 개수와 크기가 Debezium 커넥터의 메모리 소비를 크게 좌우하는데, 각 객체에 String을 하나 더하는 건 상당히 무거워질 수 있어요. 기본값은 'false'예요. | false | boolean | | includeUnknownDatatypes (postgres) | Debezium이 지원하지 않는 데이터 타입의 필드를 처리할지 정해요. 'false'(기본값)는 그 필드를 생략하고, 'true'는 구현에 따라 달라지는 binary 표현으로 필드를 변환해요. | false | boolean | | incrementalSnapshotChunkSize (postgres) | 증분 스냅샷에서 청크(chunk)의 최대 크기(문서/로우 수)예요. | 1024 | int | | incrementalSnapshotWatermarkingStrategy (postgres) | 증분 스냅샷 중 워터마킹에 쓰는 전략을 정해요. 'insert_insert'는 open 신호와 close 신호 둘 다 시그널 데이터 컬렉션에 기록하고(기본값), 'insert_delete'는 open 신호만 기록하고 close가 해당 open 신호를 지워요. | INSERT_INSERT | String | | intervalHandlingMode (postgres) | 변경 이벤트에서 INTERVAL 컬럼을 어떻게 표현할지 정해요. 'string'은 정확한 ISO 형식 문자열로 표현하고, 'numeric'(기본값)은 정확하지 않은 마이크로초 변환으로 표현해요. | numeric | String | | legacySnapshotMaxThreads (postgres) | 켜면 레거시 table-per-thread 병렬 스냅샷 알고리즘을 사용해요. false로 두면(기본값) 테이블을 청크로 나눠서 모든 스냅샷 스레드에 걸쳐 처리하기 때문에 스냅샷 동시성을 더 높일 수 있어요. | false | boolean | | lsnFlushMode (postgres) | LSN 플러시 전략을 정해요. 옵션은 이렇게 있어요: 'connector'(기본값)는 Debezium이 관리하는 LSN 플러시예요(폐기 예정인 flush.lsn.source=true를 대체). 'manual'은 외부에서 관리하는 LSN 플러시예요(폐기 예정인 flush.lsn.source=false를 대체). 'connector_and_driver'는 Debezium이 관리하는 LSN 플러시에 pgjdbc 드라이버가 서버 keepalive LSN으로 감시하지 않는(unmonitored) LSN까지 플러시하는 방식이고, 활동이 적은 데이터베이스에서 WAL이 자라는 걸 막아 줘요. | | String | | lsnFlushTimeoutAction (postgres) | LSN 플러시 타임아웃이 발생했을 때 취할 동작이에요. 옵션은 이렇게 있어요: 'fail'(기본값)은 커넥터를 실패시키고, 'warn'은 경고를 로그로 남기고 계속 처리하며, 'ignore'는 타임아웃을 무시하고 계속 처리해요. | fail | String | | lsnFlushTimeoutMs (postgres) | LSN 플러시 동작이 끝나기를 기다리는 최대 시간(밀리초)이에요. 이 시간 안에 플러시가 끝나지 않으면 lsn.flush.timeout.action에 지정한 동작을 해요. 기본값은 30초예요. | 30s | long | | maxBatchSize (postgres) | 소스 레코드 배치 하나의 최대 크기예요. 기본값은 2048이에요. | 2048 | int | | maxQueueSize (postgres) | 데이터베이스 로그에서 읽었지만 아직 기록되거나 전달되지 않은 변경 이벤트 큐의 최대 크기예요. 기본값은 8192이고, 항상 최대 배치 크기보다 커야 해요. | 8192 | int | | maxQueueSizeInBytes (postgres) | 같은 큐의 바이트 단위 최대 크기예요. 기본값은 0이고, 이는 이 기능이 꺼져 있다는 뜻이에요. | 0 | long | | memoryManagementSchemasClass (postgres) | 스키마 메타데이터용 스토리지 구현의 완전한 클래스 이름(fully-qualified class name)이에요. 이 클래스는 io.debezium.relational.TableMappingStorage를 구현해야 하고, 기본값은 인메모리 스토리지용 io.debezium.relational.ConcurrentMapTableMappingStorage예요. | io.debezium.relational.ConcurrentMapTableMappingStorage | String | | memoryManagementTablesClass (postgres) | 테이블 메타데이터용 스토리지 구현의 완전한 클래스 이름(fully-qualified class name)이에요. 이 클래스는 io.debezium.relational.TableMappingStorage를 구현해야 하고, 기본값은 인메모리 스토리지용 io.debezium.relational.ConcurrentMapTableMappingStorage예요. | io.debezium.relational.ConcurrentMapTableMappingStorage | String | | messageKeyColumns (postgres) | 메시지 키로 쓸 완전한 이름(fully-qualified)의 테이블과 컬럼(들)과 일치하는 표현식을 세미콜론으로 나열한 목록이에요. 각 표현식은 ':' 패턴과 일치해야 하고, 테이블 이름은 커넥터에 따라 (DB_NAME.TABLE_NAME)이나 (SCHEMA_NAME.TABLE_NAME)으로 정의할 수 있으며, 키 컬럼은 커스텀 키를 나타내는 컬럼의 쉼표 목록이에요. 명시적인 키 설정이 없는 테이블은 테이블의 기본 키 컬럼(들)이 메시지 키로 쓰여요. 예: dbserver1.inventory.orderlines:orderId,orderLineId;dbserver1.inventory.orders:id. | | String | | messagePrefixExcludeList (postgres) | 모니터링에서 제외할 논리 디코딩(logical decoding) 메시지 접두어와 일치하는 정규 표현식을 쉼표로 나열한 목록이에요. | | String | | messagePrefixIncludeList (postgres) | 모니터링할 논리 디코딩(logical decoding) 메시지 접두어와 일치하는 정규 표현식을 쉼표로 나열한 목록이에요. 기본값으로 모든 접두어를 모니터링해요. | | String | | notificationEnabledChannels (postgres) | 활성화할 알림(notifcation) 채널 이름의 목록이에요. | | String | | notificationSinkTopicName (postgres) | 알림을 담을 토픽의 이름이에요. 활성화된 채널 목록에 'sink'가 있으면 필수예요. | | String | | openlineageIntegrationConfigFilePath (postgres) | OpenLineage 파일 설정의 경로예요. 자세한 내용은 https://openlineage.io/docs/client/java/configuration을 참고하세요. | ./openlineage.yml | String | | openlineageIntegrationDatasetKafkaBootstrapServers (postgres) | 입력/출력 네임스페이스로 쓰이는 Kafka 부트스트랩 서버 주소예요. | | String | | openlineageIntegrationEnabled (postgres) | Debezium이 OpenLineage API로 데이터 리니지(data lineage) 메타데이터를 내보내도록 켜요. | false | boolean | | openlineageIntegrationJobDescription (postgres) | Debezium이 내보내는 job의 설명이에요. | Debezium change data capture job | String | | openlineageIntegrationJobNamespace (postgres) | Debezium이 내보내는 job의 네임스페이스예요. | | String | | openlineageIntegrationJobOwners (postgres) | Debezium이 내보내는 job의 소유자(owners)예요. 키-값 쌍을 쉼표로 나열해요. 예: k1=v1,k2=v2. | | String | | openlineageIntegrationJobTags (postgres) | Debezium이 내보내는 job의 태그예요. 키-값 쌍을 쉼표로 나열해요. 예: k1=v1,k2=v2. | | String | | pluginName (postgres) | 서버에 설치된 Postgres 논리 디코딩(logical decoding) 플러그인의 이름이에요. 지원하는 값은 'decoderbufs'와 'pgoutput'이고, 기본값은 'decoderbufs'예요. | decoderbufs | String | | pollIntervalMs (postgres) | 이벤트가 없을 때 새 변경 이벤트가 나타나기를 기다리는 시간(밀리초)이에요. 기본값은 500ms예요. | 500ms | long | | postProcessors (postgres) | 포스트 프로세서(post processor) 목록(선택)이에요. 프로세서는 '.type' 설정 옵션으로 정의하고 '' 옵션으로 설정해요. | | String | | provideTransactionMetadata (postgres) | 이벤트 카운트와 함께 트랜잭션 메타데이터 추출을 켜요. | false | boolean | | publicationAutocreateMode (postgres) | pgoutput으로 변경을 스트리밍할 때만 적용돼요. publication 생성을 어떻게 동작하게 할지 정하고, 기본값은 all_tables예요. DISABLED - 커넥터가 publication을 전혀 만들려고 하지 않아요. 사용자가 미리 publication을 만들어 뒀다고 가정해요. 시작할 때 publication이 없으면 커넥터는 예외를 던지고 멈춰요. ALL_TABLES - publication이 없으면 모든 테이블에 대한 새 publication을 만들어요. 이때 설정된 사용자가 접근 권한이 있어야 한다는 점에 유의하세요. publication이 이미 있으면 그걸 사용해요. 즉 CREATE PUBLICATION FOR ALL TABLES예요. FILTERED - publication이 없으면 현재 필터 설정(테이블/데이터베이스 include/exclude 목록 프로퍼티 참고)에 맞는 테이블들에 대한 새 publication을 만들어요. publication이 이미 있으면 그걸 사용해요. 즉 CREATE PUBLICATION FOR TABLE이에요. | all_tables | String | | publicationName (postgres) | 플러그인에서 변경을 스트리밍할 때 쓰는 Postgres 10 publication의 이름이에요. 기본값은 'dbz_publication'예요. | dbz_publication | String | | publishViaPartitionRoot (postgres) | 커넥터가 파티션 루트(partition root)를 통해 변경을 발행할지 정하는 불리언이에요. true면 파티션 루트를 통해 발행하고, false면 바로 발행해요. | false | boolean | | queryFetchSize (postgres) | 스트리밍 중에 메모리로 읽어 들일 최대 레코드 수예요. '0'이면 기본 JDBC fetch size를 사용해요. | 0 | int | | replicaIdentityAutosetValues (postgres) | pgoutput으로 변경을 스트리밍할 때만 적용돼요. 테이블 수준의 Replica Identity 값을 정해요. 이 옵션은 데이터베이스에 이미 있는 값을 덮어써요. 완전한 이름(fully-qualified)의 테이블과 그 테이블에 사용할 Replica Identity 값이 일치하는 정규 표현식을 쉼표로 나열한 목록이에요. 각 표현식은 ':' 패턴과 일치해야 하고, 테이블 이름은 (SCHEMA_NAME.TABLE_NAME)으로 정의해요. Replica Identity 값은 이렇게 있어요: DEFAULT - 기본 키 컬럼의 이전 값을 기록해요(있다면). 시스템이 아닌 테이블의 기본값이에요. INDEX index_name - 지정한 인덱스가 다루는 컬럼의 이전 값을 기록해요. 이 인덱스는 유니크해야 하고, partial이 아니어야 하고, deferrable이 아니어야 하며, NOT NULL로 표시된 컬럼만 포함해야 해요. 이 인덱스를 지우면 NOTHING과 같은 동작을 해요. FULL - 로우의 모든 컬럼에 대한 이전 값을 기록해요. NOTHING - 이전 로우에 대한 정보를 기록하지 않아요. 시스템 테이블의 기본값이에요. | | String | | retriableRestartConnectorWaitMs (postgres) | 재시도 가능한(retriable) 예외가 발생한 뒤 커넥터를 다시 시작하기 전에 기다리는 시간이에요. 기본값은 10000ms예요. | 10s | long | | schemaExcludeList (postgres) | 이벤트를 캡처하면 안 되는 스키마들이에요. | | String | | schemaHistoryInternalFileFilename (postgres) | 데이터베이스 스키마 히스토리를 기록하는 데 쓸 파일의 경로예요. | | String | | schemaIncludeList (postgres) | 이벤트를 캡처할 스키마들이에요. | | String | | schemaNameAdjustmentMode (postgres) | 커넥터가 사용하는 메시지 컨버터와의 호환성을 위해 스키마 이름을 어떻게 조정할지 정해요. 'avro'는 Avro 타입 이름에 쓸 수 없는 문자를 언더스코어로 바꿔요. 'avro_unicode'는 언더스코어나 Avro 타입 이름에 쓸 수 없는 문자를 _uxxxx 같은 대응하는 유니코드로 바꿔요. 참고: _는 Java의 백슬래시 같은 이스케이프 시퀀스예요. 'none'은 아무 조정도 하지 않아요(기본값). | none | String | | schemaRefreshMode (postgres) | 테이블의 인메모리 스키마를 새로 고치는 조건을 정해요. 'columns_diff'(기본값)가 가장 안전한 모드로, 인메모리 스키마가 항상 데이터베이스 테이블 스키마와 동기화되도록 보장해요. 'columns_diff_exclude_unchanged_toast'는 인메모리 스키마 캐시와 들어오는 메시지에서 도출한 스키마 사이에 불일치가 있을 때 캐시를 새로 고치라고 지시하지만, 변경되지 않은 TOASTable 데이터가 그 불일치를 전부 설명한다면 예외예요. 자주 업데이트되는 테이블에 TOASTed 데이터가 있고 그 데이터가 이런 업데이트에 잘 끼어들지 않는다면 이 설정은 커넥터 성능을 크게 개선할 수 있어요. 다만 TOASTable 컬럼을 테이블에서 지우면 인메모리 스키마가 낡을 수 있어요. | columns_diff | String | | signalDataCollection (postgres) | Debezium에 시그널/커맨드를 보낼 때 쓰는 데이터 컬렉션의 이름이에요. 멀티 파티션 모드 커넥터는 시그널 데이터 컬렉션을 쉼표로 구분해서 여러 개 지정할 수 있어요. 설정하지 않으면 시그널링은 꺼져 있어요. | | String | | signalEnabledChannels (postgres) | 활성화된 채널 이름의 목록이에요. source 채널이 기본으로 활성화돼요. | source | String | | signalPollIntervalMs (postgres) | 등록된 채널에서 새 시그널을 찾는 간격(밀리초)이에요. 기본값은 5초예요. | 5s | long | | skippedOperations (postgres) | 스트리밍 중에 건너뛸 연산을 쉼표로 나열한 목록이에요. 'c'는 insert/create, 'u'는 update, 'd'는 delete, 't'는 truncate를 뜻하고, 'none'은 아무것도 건너뛰지 않는다는 뜻이에요. 기본값으로는 truncate 연산만 건너뛰어요. | t | String | | slotDropOnStop (postgres) | 커넥터가 정상적으로 끝났을 때 논리 복제 슬롯(logical replication slot)을 지울지 여부예요. 기본값으로는 복제 슬롯을 유지해서, 재시작할 때 마지막으로 기록된 위치부터 이어서 진행할 수 있게 해요. | false | boolean | | slotFailover (postgres) | 페일오버 슬롯(failover slot)을 만들지 여부예요. Postgres 클러스터의 프라이머리 서버에 접속할 때, 버전 17 이상에서만 지원해요. 지정하지 않거나 Postgres 17 프라이머리에 접속하지 않으면 페일오버 슬롯을 만들지 않아요. | false | boolean | | slotMaxRetries (postgres) | 복제 슬롯 접속 시도가 실패했을 때 재시도하는 횟수예요. | 6 | int | | slotName (postgres) | 플러그인에서 변경을 스트리밍하려고 만든 Postgres 논리 디코딩(logical decoding) 슬롯의 이름이에요. 기본값은 'debezium이에요. | debezium | String | | slotRetryDelayMs (postgres) | 커넥터가 복제 슬롯 접속에 실패했을 때 재시도 사이에 기다리는 시간(밀리초)이에요. 기본값은 10초(10,000ms)예요. | 10s | long | | slotStreamParams (postgres) | 논리 디코딩(logical decoding) 플러그인이 쓰는 선택적 파라미터예요. 세미콜론으로 구분해요. 예: 'add-tables=public.table,public.table2;include-lsn=true'. | | String | | snapshotDelayMs (postgres) | 스냅샷이 시작되기 전의 지연 시간(밀리초)이에요. 기본값은 0ms예요. | 0ms | long | | snapshotFetchSize (postgres) | 스냅샷을 수행하는 동안 메모리로 읽어 들일 최대 레코드 수예요. | | int | | snapshotIncludeCollectionList (postgres) | 커넥터를 만들거나 재시작할 때 스냅샷을 찍어야 할 테이블/컬렉션 목록을 지정하려면 이 설정을 반드시 지정해야 해요. | | String | | snapshotIsolationMode (postgres) | 어떤 트랜잭션 격리 수준을 쓸지 조절해요. 기본값은 'serializable'로, serializable 격리 수준을 쓴다는 뜻이에요. 'repeatable_read'를 지정하면 커넥터가 초기 스냅샷을 REPEATABLE READ 격리 수준으로 수행해요. 'read_committed'를 지정하면 READ COMMITTED 격리 수준으로 수행해요. 'read_uncommitted'를 지정하면 READ UNCOMMITTED 격리 수준으로 수행해요. | serializable | String | | snapshotLockingMode (postgres) | 스키마 스냅샷을 수행하는 동안 커넥터가 테이블 잠금을 어떻게 쥐고 있을지 조절해요. 'shared'는 커넥터가 데이터베이스 스키마와 다른 메타데이터를 읽는 스냅샷 초기 구간에만 배타적 테이블 접근을 막는 테이블 잠금을 쥐고 있게 해요. 스냅샷의 나머지 작업은 각 테이블에서 모든 로우를 select하는 건데, 이때는 잠금이 필요 없는 flashback query를 사용해요. 하지만 어떤 경우엔 잠금을 아예 피하는 게 좋을 수도 있는데, 'none'을 지정하면 그렇게 할 수 있어요. 이 모드는 스냅샷을 찍는 동안 스키마 변경이 일어나지 않을 때만 안전해요. | none | String | | snapshotLockingModeCustomName (postgres) | 'snapshot.locking.mode'가 custom으로 설정되어 있으면, 'name()' 메서드가 제공하는 커스텀 구현의 이름을 지정하도록 이 설정을 반드시 지정해야 해요. 구현체는 'SnapshotterLocking' 인터페이스를 구현해야 하고, 스키마 스냅샷 중 테이블을 어떻게 잠글지 정하려고 호출돼요. | | String | | snapshotLockTimeoutMs (postgres) | 스냅샷 시작 시 테이블 잠금을 기다리는 최대 밀리초예요. 이 시간 안에 잠금을 얻지 못하면 스냅샷이 중단돼요. 기본값은 10초예요. | 10s | long | | snapshotMaxThreads (postgres) | 스냅샷을 수행할 때 쓰는 최대 스레드 수예요. 기본값은 1이에요. | 1 | int | | snapshotMaxThreadsMultiplier (postgres) | 테이블당 스냅샷 청크 개수를 늘리는 데 쓰는 계수예요. 기본 동작은 'row_count/snapshot.max.threads'로 로우당 청크 수를 계산해요. 큰 테이블에는 이게 최선이 아닐 수 있는데, multiplier를 쓰면 'row_count/(snapshot.max.threads snapshot.max.threads.multiplier)'로 청크 개수를 늘리도록 공식이 조정돼요. | 1 | int | | snapshotMode (postgres) | 커넥터 시작 시 스냅샷을 실행하는 기준이에요. 다음 스냅샷 옵션 중 하나를 고르세요: 'always': 커넥터가 시작할 때마다 스냅샷을 실행해요. 스냅샷이 끝나면 트랜잭션 로그에서 변경을 스트리밍하기 시작해요. 'initial'(기본값): 커넥터가 논리 서버 이름에 대한 오프셋을 찾지 못하면, 설정된 테이블의 현재 전체 상태를 담는 스냅샷을 실행해요. 스냅샷이 끝나면 트랜잭션 로그에서 변경을 스트리밍하기 시작해요. 'initial_only': 'initial' 옵션처럼 스냅샷을 수행하지만, 스냅샷이 끝나면 커넥터가 멈추고 트랜잭션 로그에서 변경을 스트리밍하지 않아요. 'never': 커넥터가 스냅샷을 실행하지 않아요. 처음 시작하면 즉시 트랜잭션 로그의 처음부터 읽기 시작해요. 'exported': 이 옵션은 폐기 예정이에요. 대신 'initial'을 사용하세요. 'custom': 커넥터가 스냅샷을 수행하는 방법을 지정할 커스텀 클래스를 로드해요. 자세한 내용은 PostgreSQL 커넥터 문서의 Custom snapshotter SPI를 참고하세요. | initial | String | | snapshotModeConfigurationBasedSnapshotData (postgres) | 'snapshot.mode'가 configuration_based로 설정되어 있으면, 데이터를 스냅샷할지 말지를 지정할 수 있게 해 줘요. | false | boolean | | snapshotModeConfigurationBasedSnapshotOnDataError (postgres) | 'snapshot.mode'가 configuration_based로 설정되어 있으면, 오류가 났을 때 데이터를 스냅샷할지 말지를 지정할 수 있게 해 줘요. | false | boolean | | snapshotModeConfigurationBasedSnapshotOnSchemaError (postgres) | 'snapshot.mode'가 configuration_based로 설정되어 있으면, 오류가 났을 때 스키마를 스냅샷할지 말지를 지정할 수 있게 해 줘요. | false | boolean | | snapshotModeConfigurationBasedSnapshotSchema (postgres) | 'snapshot.mode'가 configuration_based로 설정되어 있으면, 스키마를 스냅샷할지 말지를 지정할 수 있게 해 줘요. | false | boolean | | snapshotModeConfigurationBasedStartStream (postgres) | 'snapshot.mode'가 configuration_based로 설정되어 있으면, 스냅샷 이후에 스트림을 시작할지 말지를 지정할 수 있게 해 줘요. | false | boolean | | snapshotModeCustomName (postgres) | 'snapshot.mode'가 custom으로 설정되어 있으면, 'name()' 메서드가 제공하는 커스텀 구현의 이름을 지정하도록 이 설정을 반드시 지정해야 해요. 구현체는 'Snapshotter' 인터페이스를 구현해야 하고, 앱이 부팅될 때마다 호출돼서 스냅샷을 할지 정해요. | | String | | snapshotQueryMode (postgres) | 스냅샷 동안 사용할 쿼리를 조절해요. | select_all | String | | snapshotQueryModeCustomName (postgres) | 'snapshot.query.mode'가 custom으로 설정되어 있으면, 'name()' 메서드가 제공하는 커스텀 구현의 이름을 지정하도록 이 설정을 반드시 지정해야 해요. 구현체는 'SnapshotterQuery' 인터페이스를 구현해야 하고, 스냅샷 동안 쿼리를 어떻게 만들지 정하려고 호출돼요. | | String | | snapshotSelectStatementOverrides (postgres) | 이 프로퍼티는 완전한 이름(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 (postgres) | 초기 스냅샷에서 테이블을 처리하는 순서를 조절해요. descending 값을 주면 테이블을 로우 수 내림차순으로 정렬해요. ascending 값을 주면 로우 수 오름차순으로 정렬해요. disabled(기본값)는 로우 수 정렬을 꺼요. | disabled | String | | sourceinfoStructMaker (postgres) | SourceInfo 스키마와 struct를 돌려주는 SourceInfoStructMaker 클래스의 이름이에요. | io.debezium.connector.postgresql.PostgresSourceInfoStructMaker | String | | statisticsMetricsEnabled (postgres) | 레코드 처리 지연 시간 같은 여러 통계와 quantiles 같은 파생 데이터를 수집하게 해요. 기본값으로 통계 수집은 켜져 있어요. | true | boolean | | statusUpdateIntervalMs (postgres) | 복제 커넥션 상태 업데이트를 서버로 보내는 빈도(밀리초)예요. 기본값은 10초(10,000ms)예요. | 10s | int | | streamingDelayMs (postgres) | 스냅샷이 끝나고 스트리밍이 시작되기까지의 지연 시간(밀리초)이에요. 기본값은 0ms예요. | 0ms | long | | tableExcludeList (postgres) | 모니터링에서 제외할 테이블의 완전한 이름(fully-qualified name)과 일치하는 정규 표현식을 쉼표로 나열한 목록이에요. | | String | | tableIgnoreBuiltin (postgres) | 내장(built-in) 테이블을 무시할지 여부를 지정하는 플래그예요. | true | boolean | | tableIncludeList (postgres) | 변경을 캡처할 테이블들이에요. | | String | | timePrecisionMode (postgres) | 시간, 날짜, 타임스탬프는 여러 정밀도로 표현할 수 있어요. 'adaptive'(기본값)는 시간·날짜·타임스탬프 값의 정밀도를 데이터베이스 컬럼의 정밀도에 맞춰요. 'adaptive_time_microseconds'는 'adaptive' 모드와 비슷하지만 TIME 필드는 항상 마이크로초 정밀도를 써요. 'connect'는 데이터베이스 컬럼의 정밀도와 무관하게 항상 밀리초 정밀도의 Kafka Connect 내장 Time·Date·Timestamp 표현으로 표현해요. 'isostring'은 시간·날짜·타임스탬프 값을 UTC 시간대의 ISO-8601 형식 문자열로 표현해요. 'microseconds'는 항상 마이크로초 정밀도로 표현해요. 'nanoseconds'는 항상 나노초 정밀도로 표현해요. | adaptive | String | | tombstonesOnDelete (postgres) | delete 연산을 delete 이벤트와 그 뒤의 tombstone 이벤트로 표현할지(true), delete 이벤트만으로 표현할지(false) 여부예요. tombstone 이벤트를 내보내면(기본 동작) 소스 레코드가 지워졌을 때 Kafka가 해당 키와 관련된 모든 이벤트를 완전히 삭제할 수 있어요. | false | boolean | | topicNamingStrategy (postgres) | 데이터 변경, 스키마 변경, 트랜잭션, 하트비트 이벤트 등의 토픽 이름을 정하는 데 쓸 TopicNamingStrategy 클래스의 이름이에요. | io.debezium.schema.SchemaTopicNamingStrategy | String | | topicPrefix (postgres) | 필수 특정 데이터베이스 서버/클러스터가 변경을 캡처하고 있음을 식별하고 네임스페이스를 제공하는 토픽 접두어예요. 토픽 접두어는 다른 모든 커넥터와 구별되게 유니크해야 해요. 이 커넥터가 내보내는 이벤트를 받는 모든 Kafka 토픽 이름의 접두어로 쓰이기 때문이에요. 영숫자, 하이픈, 점, 언더스코어만 받아들여야 해요. | | String | | transactionMetadataFactory (postgres) | 트랜잭션 컨텍스트와 트랜잭션 struct/스키마를 만드는 클래스예요. | io.debezium.pipeline.txmetadata.DefaultTransactionMetadataFactory | String | | unavailableValuePlaceholder (postgres) | 원래 값이 데이터베이스가 제공하지 않는 toasted 값임을 나타내는 상수를 정해요. 'hex:' 접두어로 시작하면 나머지 문자열이 16진수로 인코딩된 옥텟을 뜻하는 걸로 기대해요. | __debezium_unavailable_value | String | | xminFetchIntervalMs (postgres) | 복제 슬롯에서 xmin을 얼마나 자주(밀리초) 가져올지 정해요. 이 xmin 값은 슬롯이 노출하는 값으로, 새 복제 슬롯이 어디서 시작할 수 있는지의 하한(lower bound)이 돼요. 값이 작을수록 이 값이 현재의 '참' 값일 가능성이 높지만 성능 비용은 커져요. 값이 클수록 현재의 '참' 값일 가능성은 낮아지지만 성능 부담은 줄어들어요. 기본값은 0ms로, xmin 추적을 꺼요. | 0ms | long |

메시지 헤더 (Message Headers)

Debezium PostgreSQL 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

설정에 대해 더 자세히 알아보려면 임베디드 엔진 프로퍼티와 PostgreSQL 커넥터 프로퍼티를 참고하세요.

사용법 (Usage)

메시지 바디 (Message body)

메시지 바디가 null이 아니라면(톰브스톤의 경우 null이에요) 이벤트가 발생한 뒤의 로우 상태를 담고 있어요. 두 가지 형식 중 하나예요:

  • Struct

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

자세한 내용은 아래를 참고하세요.

예제 (Examples)

이벤트 컨슈밍 (Consuming events)

PostgreSQL 커넥터의 Debezium 이벤트를 받아 보는 기본 라우트예요.

  • Java

  • XML

  • YAML

from("debezium-postgres: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}");

<route>
  <from uri="debezium-postgres: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}"/>
</route>

- route:
    from:
      uri: debezium-postgres: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}"

기본으로 컴포넌트는 바디와 CamelDebeziumBefore 헤더에 이벤트를 Struct 데이터 타입으로 내보내요. 필요할 때 스키마 정보를 볼 수 있게 하기 위해서예요. 하지만 이 컴포넌트에는 기본 출력 타입인 Struct를 Map으로 변환하는 Type Converter도 들어 있어요. Map 데이터 타입으로 바로 동작하는 것이 많은 Camel의 풍부한 Data Format 타입을 활용하기 위해서죠. 이걸 쓰려면 메시지에 접근할 때 Map.class 타입을 지정하거나(예: exchange.getIn().getBody(Map.class)), 라우트 빌더에서 from 문 뒤에 .convertBodyTo(Map.class)를 추가해서 바디를 항상 Map으로 변환할 수도 있어요.

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

Java 전용: processor에서 Struct 스키마 접근하기

from("debezium-postgres:[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을 얇게 감싼 래퍼예요. 따라서 프로덕션에서 이 컴포넌트를 사용하기 전에 Debezium이 어떻게 동작하는지, 그리고 설정이 기대한 동작을 어떻게 반영하는지 이해해야 해요. 특히 장애 처리 부분이 그래요.

더 알아보기 (Learn more)

장애 처리와 중복 이벤트 동작은 본문 앞쪽의 참고 블록과 handling failures 섹션을 먼저 확인하세요. 실제 라우트 예시는 위의 예제 (Examples) 섹션을 참고하고, 의존성 좌표는 camel-debezium-postgres 아티팩트 그대로 쓰면 돼요.

함께 읽는 실무 이야기

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