Kafka Connect 설정

Kafka Connect 설정 (Kafka Connect Configs)

Kafka Connect 프레임워크(워커·커넥터·소스/싱크 커넥터)의 설정 항목들을 정리한 문서예요. 데이터를 외부 시스템과 주고받는 Connect를 운영할 때 워커 그룹, 저장 토픽, 컨버터, 오류 처리, REST API 등을 여기서 설정해요.

출처: 문서

참고: 공통 보안 항목(ssl.*, sasl.*, sasl.oauthbearer.*, security.protocol) 등은 다른 Kafka 클라이언트 설정 문서와 동일해요. 여기서는 Connect 프레임워크 특유의 설정을 중심으로 설명할게요.

본문

아래는 Kafka Connect 프레임워크 설정 항목들이에요.

• bootstrap.servers — Kafka 클러스터에 초기 연결을 설정하는 데 쓰는 host/port 쌍 목록. 클라이언트가 이 목록으로 부트스트랩해 전체 Kafka 브로커 집합을 발견해요. 형식은 host1:port1,host2:port2,.... 유형: list / 중요도: high
• config.storage.topic — 커넥터 구성을 저장하는 Kafka 토픽 이름. 유형: string / 중요도: high
• group.id — 이 워커가 속한 Connect 클러스터 그룹을 식별하는 고유 문자열. 유형: string / 중요도: high
• key.converter — Kafka Connect 형식과 Kafka에 쓰이는 직렬화 형식 사이를 변환하는 컨버터 클래스. 메시지 키의 형식을 제어하며 커넥터와 독립적이라 어떤 커넥터든 어떤 직렬화 형식과도 동작해요. 일반적인 형식에는 JSON·Avro가 있어요. 유형: class / 중요도: high
• offset.storage.topic — 소스 커넥터 오프셋을 저장하는 Kafka 토픽 이름. 유형: string / 중요도: high
• status.storage.topic — 커넥터와 작업의 상태를 저장하는 Kafka 토픽 이름. 유형: string / 중요도: high
• value.converter — Kafka Connect 형식과 Kafka에 쓰이는 직렬화 형식 사이를 변환하는 컨버터 클래스. 메시지 값의 형식을 제어해요. 유형: class / 중요도: high
• exactly.once.source.support — 클러스터의 소스 커넥터에 exactly-once 지원을 활성화할지 여부. 트랜잭션으로 소스 레코드와 소스 오프셋을 쓰고, 새 작업 세대를 시작하기 전에 이전 작업 세대를 능동적으로 펜싱(fencing)해요. 새 클러스터에서 활성화하려면 enabled로, 기존 클러스터에서는 먼저 모든 워커에서 preparing으로 설정한 뒤 enabled로 설정해요. 두 변경 모두 롤링 업그레이드 사용 가능. 자세한 내용은 exactly-once source support 문서 참고.
  • 유형: string / 기본값: disabled / 유효 값: (대소문자 무시) [DISABLED, ENABLED, PREPARING] / 중요도: high
• heartbeat.interval.ms — 그룹 관리 사용 시 그룹 코디네이터로 보내는 하트비트 사이의 예상 시간. 워커 세션을 활성으로 유지하고 리밸런스를 돕는 데 사용돼요. 값은 session.timeout.ms보다 낮아야 하고 보통 그 값의 1/3 이하로 설정하는 게 좋아요.
  • 유형: int / 기본값: 3000 (3초) / 중요도: high
• rebalance.timeout.ms — 리밸런스가 시작된 후 각 워커가 그룹에 조인하기 위해 허용되는 최대 시간. 본질적으로 모든 작업이 대기 데이터를 플러시하고 오프셋을 커밋하는 데 필요한 시간의 한도예요. 이 타임아웃을 초과하면 워커가 그룹에서 제거되어 오프셋 커밋 실패를 일으킬 수 있어요.
  • 유형: int / 기본값: 60000 (1분) / 중요도: high
• session.timeout.ms — 워커 장애를 감지하는 타임아웃. 워커는 주기적 하트비트를 보내 활성 상태를 알려요. 세션 타임아웃 만료 전에 하트비트를 받지 못하면 브로커가 워커를 그룹에서 제거하고 리밸런스를 시작해요. 값은 브로커 구성의 group.min.session.timeout.ms·group.max.session.timeout.ms 범위 안이어야 해요.
  • 유형: int / 기본값: 10000 (10초) / 중요도: high
• connector.client.config.override.policy — ConnectorClientConfigOverridePolicy 구현의 클래스 이름 또는 별칭. 커넥터가 재정의할 수 있는 클라이언트 구성을 정의해요. 기본 정책은 All로, 커넥터 구성이 모든 클라이언트 속성을 재정의할 수 있음을 뜻해요. 다른 정책: Allowlist(connector.client.config.override.allowlist로 허용 구성 지정), None(재정의 금지), Principal(현재 비권장, 클라이언트 주체만 재정의 허용).
  • 유형: string / 기본값: All / 중요도: medium
• worker.sync.timeout.ms — 워커가 다른 워커와 동기화가 어긋나 구성을 재동기화해야 할 때, 포기하고 그룹을 떠나 백오프 후 다시 조인하기 전에 이 시간만큼 기다려요.
  • 유형: int / 기본값: 3000 (3초) / 중요도: medium
• worker.unsync.backoff.ms — 워커가 다른 워커와 동기화가 어긋나 worker.sync.timeout.ms 안에 따라잡지 못하면, 다시 조인하기 전에 Connect 클러스터를 이 시간 동안 떠나 있어요.
  • 유형: int / 기본값: 300000 (5분) / 중요도: medium
• ssl.* / sasl.* / sasl.oauthbearer.* — 클라이언트 보안 설정(다른 클라이언트 설정 문서와 동일). ssl.key.password, ssl.keystore.*, ssl.truststore.*, ssl.enabled.protocols, ssl.protocol, ssl.provider, sasl.mechanism, sasl.jaas.config, sasl.login.*, ssl.cipher.suites 등의 기본값은 Admin/Producer/Consumer 설정 페이지와 동일해요.
• access.control.allow.methods — Access-Control-Allow-Methods 헤더를 설정해 교차 출처(cross origin) 요청에 지원되는 메서드를 설정해요. 기본값은 GET, POST, HEAD를 허용해요. 유형: string / 기본값: "" / 중요도: low
• access.control.allow.origin — REST API 요청에 Access-Control-Allow-Origin 헤더로 설정할 값. 교차 출처 접근을 활성화하려면 API 접근을 허용해야 하는 애플리케이션의 도메인이나 *(모든 도메인 허용)로 설정해요. 기본값은 REST API의 도메인에서만 접근을 허용해요. 유형: string / 기본값: "" / 중요도: low
• admin.listeners — Admin REST API가 수신할 쉼표로 구분된 URI 목록. 지원 프로토콜은 HTTP·HTTPS. 빈 문자열이면 이 기능을 비활성화해요. 기본 동작은 일반 리스너(listeners 속성으로 지정)를 사용해요. 예: http://localhost:8080,https://localhost:8443. 유형: list / 기본값: null / 중요도: low
• config.storage.replication.factor — 구성 저장 토픽 생성 시 사용하는 복제 팩터. 유형: short / 기본값: 3 / 중요도: low
• connect.protocol — Kafka Connect 프로토콜의 호환성 모드. 유형: string / 기본값: sessioned / 유효 값: [eager, compatible, sessioned] / 중요도: low
• header.converter — Kafka Connect 형식과 Kafka 직렬화 형식 사이의 헤더 값을 변환하는 HeaderConverter 클래스. 기본적으로 SimpleHeaderConverter가 헤더 값을 문자열로 직렬화하고 스키마를 추론해 역직렬화해요. 유형: class / 기본값: org.apache.kafka.connect.storage.SimpleHeaderConverter / 중요도: low
• inter.worker.key.generation.algorithm — 내부 요청 키 생성에 사용할 알고리즘. HmacSHA256을 지원하는 JVM에서는 기본값 HmacSHA256이 사용되고, 다른 JVM에서는 기본값이 없어 워커 구성에 수동으로 지정해야 해요. 유형: string / 기본값: HmacSHA256 / 중요도: low
• inter.worker.key.size — 내부 요청 서명에 사용할 키 크기(비트). null이면 키 생성 알고리즘의 기본 키 크기를 사용해요. 유형: int / 기본값: null / 중요도: low
• inter.worker.key.ttl.ms — 내부 요청 검증에 사용되는 생성된 세션 키의 TTL(밀리초). 유형: int / 기본값: 3600000 (1시간) / 중요도: low
• inter.worker.signature.algorithm — 내부 요청 서명에 사용할 알고리즘. HmacSHA256을 지원하는 JVM에서는 기본값 HmacSHA256이 사용되고 다른 JVM에서는 수동 지정 필요. 유형: string / 기본값: HmacSHA256 / 중요도: low
• inter.worker.verification.algorithms — 내부 요청 검증에 허용되는 알고리즘 목록. inter.worker.signature.algorithm에 사용된 알고리즘을 포함해야 해요. 제공하는 JVM에서는 기본값 [HmacSHA256] 사용. 유형: list / 기본값: HmacSHA256 / 중요도: low
• listeners — REST API가 수신할 쉼표로 구분된 URI 목록. 지원 프로토콜은 HTTP·HTTPS. 호스트 이름을 0.0.0.0으로 지정하면 모든 인터페이스에 바인딩하고, 비우면 기본 인터페이스에 바인딩해요. 유형: list / 기본값: http://:8083 / 중요도: low
• offset.flush.interval.ms — 작업(task)의 오프셋 커밋을 시도하는 간격. 유형: long / 기본값: 60000 (1분) / 중요도: low
• offset.flush.timeout.ms — 프로세스를 취소하고 오프셋 데이터를 향후 시도에서 커밋하도록 복원하기 전에 레코드가 플러시되고 파티션 오프셋 데이터가 오프셋 저장소에 커밋되기를 기다리는 최대 밀리초. exactly-once 지원으로 실행되는 소스 커넥터에는 효과가 없어요. 유형: long / 기본값: 5000 (5초) / 중요도: low
• offset.storage.partitions — 오프셋 저장 토픽 생성 시 사용하는 파티션 수. 유형: int / 기본값: 25 / 중요도: low
• offset.storage.replication.factor — 오프셋 저장 토픽 생성 시 사용하는 복제 팩터. 유형: short / 기본값: 3 / 중요도: low
• plugin.discovery — 클래스패스와 plugin.path 구성에 있는 플러그인을 발견하는 방법. only_scan(리플렉션만, ServiceLoader로 발견 못 하는 플러그인은 워커 시작에 영향 없음), hybrid_warn(리플렉션+ServiceLoader, ServiceLoader로 못 찾으면 경고), hybrid_fail(마찬가지지만 못 찾으면 시작 실패), service_load(ServiceLoader만, 다른 모드보다 시작이 빠르지만 ServiceLoader로 못 찾는 플러그인은 사용 불가할 수 있음). 유형: string / 기본값: hybrid_warn / 유효 값: (대소문자 무시) [ONLY_SCAN, SERVICE_LOAD, HYBRID_WARN, HYBRID_FAIL] / 중요도: low
• plugin.path — 플러그인(커넥터·컨버터·변환)이 있는 쉼표로 구분된 경로 목록. 최상위 디렉토리로 구성되며, 즉시 플러그인·의존성 jar가 있는 디렉토리, 플러그인이 있는 uber-jar, 플러그인 패키지 디렉토리 구조를 즉시 포함하는 디렉토리 등을 포함해요. 심볼릭 링크는 따라가요. 이 속성에는 config provider 변수를 쓰지 마세요 — 원시 경로가 config provider 초기화 전에 워커 스캐너가 사용하거든요. 유형: list / 기본값: null / 중요도: low
• response.http.headers.config — REST API HTTP 응답 헤더 규칙. 유형: string / 기본값: "" / 중요도: low
• rest.advertised.host.name — 설정하면 다른 워커가 연결하도록 알려줄 호스트 이름. 유형: string / 기본값: null / 중요도: low
• rest.advertised.listener — 다른 워커가 사용하도록 알려줄 광고 리스너(HTTP·HTTPS). 유형: string / 기본값: null / 중요도: low
• rest.advertised.port — 설정하면 다른 워커가 연결하도록 알려줄 포트. 유형: int / 기본값: null / 중요도: low
• rest.extension.classes — ConnectRestExtension 클래스의 쉼표로 구분된 이름. 지정된 순서대로 로드·호출돼요. Connect의 REST API에 필터 같은 사용자 정의 리소스를 주입할 수 있게 해줘요. 유형: list / 기본값: "" / 중요도: low
• scheduled.rebalance.max.delay.ms — 워커 그룹에서 하나 이상의 떠난 워커의 반환을 기다리기 위해 리밸런스·재할당을 스케줄링할 최대 지연. 이 기간 동안 떠난 워커의 커넥터·작업은 할당되지 않은 채로 남아요. 유형: int / 기본값: 300000 (5분) / 중요도: low
• ssl.client.auth — Kafka 브로커가 클라이언트 인증을 요청하도록 구성. required면 클라이언트 인증이 필요, requested면 선택(클라이언트가 자신의 인증 정보를 제공하지 않을 수 있음), none이면 필요 없음. 유형: string / 기본값: none / 유효 값: [required, requested, none] / 중요도: low
• status.storage.partitions — 상태 저장 토픽 생성 시 사용하는 파티션 수. 유형: int / 기본값: 5 / 중요도: low
• status.storage.replication.factor — 상태 저장 토픽 생성 시 사용하는 복제 팩터. 유형: short / 기본값: 3 / 중요도: low
• task.shutdown.graceful.timeout.ms — 작업이 우아하게 종료되기를 기다리는 시간. 총 시간이며 작업별 시간이 아니에요. 모든 작업에 종료가 트리거된 뒤 순차적으로 기다려요. 유형: long / 기본값: 5000 (5초) / 중요도: low
• topic.creation.enable — 소스 커넥터가 topic.creation. 속성으로 구성됐을 때, 소스 커넥터가 사용하는 토픽의 자동 생성을 허용할지 여부. 각 작업은 admin 클라이언트로 토픽을 만들며 Kafka 브로커의 자동 생성에 의존하지 않아요. 유형: boolean / 기본값: true / 중요도: low
• topic.tracking.allow.reset — true면 사용자가 커넥터별 활성 토픽 집합을 재설정하는 요청을 허용해요. 유형: boolean / 기본값: true / 중요도: low
• topic.tracking.enable — 런타임 중 커넥터별 활성 토픽 집합 추적을 활성화할지 여부. 유형: boolean / 기본값: true / 중요도: low

소스 커넥터 설정 (Source Connector Configs)

아래는 소스 커넥터 설정이에요. name(전역 고유), connector.class(Connector 하위 클래스), connector.plugin.version, tasks.max(최대 작업 수, 기본 1), tasks.max.enforce(비권장), key.converter/value.converter/header.converter(+plugin.version), config.action.reload(외부 구성 변경 시 restart/none), transforms/predicates, errors.retry.timeout/errors.retry.delay.max.ms/errors.tolerance/errors.log.enable/errors.log.include.messages 같은 공통 항목은 워커 설정과 동일해요.

• topic.creation.groups — 소스 커넥터가 만드는 토픽에 대한 구성 그룹. 유형: list / 기본값: "" / 중요도: low
• exactly.once.support — 허용 값은 requested, required. required로 설정하면 커넥터가 주어진 구성으로 exactly-once 의미를 제공할 수 있는지 확인하는 사전 검사(preflight check)를 강제해요. 어떤 커넥터는 exactly-once를 제공해도 Connect에 지원을 알리지 않을 수 있어, 그 경우 커넥터 문서를 주의 깊게 확인하고 requested로 설정해야 해요. required인데 사전 검증을 수행하는 워커가 소스 커넥터용 exactly-once 지원을 활성화하지 않았다면 커넥터 생성·검증 요청이 실패해요. 유형: string / 기본값: requested / 유효 값: (대소문자 무시) [REQUIRED, REQUESTED] / 중요도: medium
• transaction.boundary — 허용 값: poll, interval, connector. poll이면 커넥터의 각 작업이 Connect에 제공하는 레코드 배치마다 새 프로듀서 트랜잭션이 시작·커밋돼요. connector는 커넥터 정의 트랜잭션 경계에 의존하는데, 모든 커넥터가 자체 경계를 정의할 수 있는 건 아니고 그런 경우 이 값으로 인스턴스화를 시도하면 실패해요. interval은 사용자 정의 시간 간격이 지난 후에만 트랜잭션을 커밋해요. 유형: string / 기본값: poll / 유효 값: (대소문자 무시) [INTERVAL, POLL, CONNECTOR] / 중요도: medium
• transaction.boundary.interval.ms — transaction.boundary가 interval일 때 커넥터 작업의 프로듀서 트랜잭션 커밋 간격. 설정하지 않으면 워커 레벨 offset.flush.interval.ms 값을 기본으로 사용해요. 다른 transaction.boundary를 지정하면 효과가 없어요. 유형: long / 기본값: null / 중요도: low
• offsets.storage.topic — 이 커넥터에 사용할 별도 오프셋 토픽 이름. 비우거나 지정하지 않으면 워커의 전역 오프셋 토픽 이름을 사용해요. 지정하면 토픽이 없으면 생성돼요. 분산 모드에서만 적용되고 standalone 모드에서는 효과가 없어요. 유형: string / 기본값: null / 중요도: low

싱크 커넥터 설정 (Sink Connector Configs)

아래는 싱크 커넥터 설정이에요. 공통 항목(name, connector.class, tasks.max, 컨버터, errors.*, transforms/predicates 등)은 위와 동일해요.

• topics — 소비할 토픽 목록(쉼표로 구분). 유형: list / 기본값: "" / 중요도: high
• topics.regex — 소비할 토픽을 주는 정규식. 내부적으로 java.util.regex.Pattern으로 컴파일돼요. topics와 topics.regex 중 하나만 지정해야 해요. 유형: string / 기본값: "" / 유효 값: 유효한 정규식 / 중요도: high
• errors.deadletterqueue.topic.name — 이 싱크 커넥터나 그 변환·컨버터가 처리 중 오류를 일으킨 메시지에 대해 DLQ(데드 레터 큐)로 사용할 토픽 이름. 기본값은 비어 있어 DLQ에 기록되는 메시지가 없어요. 유형: string / 기본값: "" / 중요도: medium
• errors.deadletterqueue.topic.replication.factor — DLQ 토픽이 없을 때 생성 시 사용하는 복제 팩터. 유형: short / 기본값: 3 / 중요도: medium
• errors.deadletterqueue.context.headers.enable — true면 오류 컨텍스트가 담긴 헤더를 DLQ에 쓰이는 메시지에 추가해요. 원본 레코드의 헤더와 충돌을 피하기 위해 모든 오류 컨텍스트 헤더 키는 __connect.errors.로 시작해요. 유형: boolean / 기본값: false / 중요도: medium

더 알아보기 (Learn more)

• 워커 그룹은 group.id로 묶이고, config.storage.topic·offset.storage.topic·status.storage.topic이 커넥터 상태를 저장하는 내부 토픽이에요.
• key.converter/value.converter로 직렬화 형식(JSON·Avro 등)을 결정하고, plugin.path로 커넥터·변환·컨버터 플러그인을 로드해요.
• 소스 커넥터의 exactly-once 지원은 exactly.once.source.support(클러스터) + exactly.once.support·transaction.boundary(커넥터)로 제어해요.