Kafka Streams 설정

Kafka Streams 설정 (Kafka Streams Configs)

Kafka Streams 클라이언트 라이브러리를 쓸 때 쓰는 설정 항목들을 정리한 문서예요. 애플리케이션 식별자, 직렬화(serde), 처리 보증, 상태 저장소(rocksDB) 설정까지 Streams 앱 동작을 결정하는 값들이 대부분 여기에 들어 있어요.

출처: 문서

본문

아래는 Kafka Streams 클라이언트 라이브러리 설정 항목들이에요.

• application.id — 스트림 처리 애플리케이션의 식별자. Kafka 클러스터 안에서 고유해야 해요. 1) 기본 client-id 접두사, 2) 멤버십 관리를 위한 group-id, 3) 체인지로그(changelog) 토픽 접두사로 사용돼요.
  • 유형: string / 기본값: — / 중요도: high
• bootstrap.servers — Kafka 클러스터에 초기 연결을 설정하는 데 쓰는 host/port 쌍 목록. 클라이언트가 이 목록으로 부트스트랩해 전체 Kafka 브로커 집합을 발견해요. 목록의 서버 순서는 상관없지만, 어떤 서버가 다운되더라도 복원력을 위해 두 개 이상의 서버를 포함하는 걸 권장해요. 이 목록에 브로커 전체가 포함될 필요는 없어요 — Kafka 클라이언트가 클러스터 연결을 효율적으로 자동 관리·갱신하거든요. 형식은 host1:port1,host2:port2,...이어야 해요.
  • 유형: list / 기본값: — / 중요도: high
• ensure.explicit.internal.resource.naming — 토폴로지의 모든 내부 리소스(체인지로그·repartition 토픽 같은 내부 토픽과 그와 연결된 상태 저장소 포함)에 대해 명시적 이름 지정을 강제할지 여부. 활성화하면 내부 리소스에 자동 생성된 이름이 있으면 애플리케이션이 시작을 거부해요.
  • 유형: boolean / 기본값: false / 중요도: high
• num.standby.replicas — 각 작업(task)의 대기(standby) 복제본 수.
  • 유형: int / 기본값: 0 / 중요도: high
• processing.exception.handler.global.enabled — 전역 저장소(global store)/KTable 처리 중에 설정된 processing.exception.handler를 사용할지 여부. 기본은 비활성화예요. 이 구성은 Kafka Streams 5.0에서 제거될 예정인데, 그때 전역 예외 처리가 기본으로 활성화돼요.
  • 유형: boolean / 기본값: false / 중요도: high
• state.dir — 상태 저장소의 디렉토리 위치. 같은 기본 파일시스템을 공유하는 각 streams 인스턴스에 대해 이 경로는 고유해야 해요. 참고로 설정하지 않으면 System.getProperty("java.io.tmpdir")로 계산되므로 환경마다 기본 위치가 달라질 수 있어요.
  • 유형: string / 기본값: ${java.io.tmpdir} / 중요도: high
• acceptable.recovery.lag — 클라이언트가 활성 작업(active task) 할당을 받기에 충분히 따라잡은 것으로 간주되는 최대 허용 지연(따라잡을 오프셋 수). 할당 후에도 처리를 시작하기 전에 나머지 체인지로그를 복원해요. 리밸런스 중 처리 중단을 피하려면, 이 구성이 주어진 워크로드에 대해 1분 미만의 복구 시간에 해당해야 해요. 최소 0이어야 해요.
  • 유형: long / 기본값: 10000 / 유효 값: [0,...] / 중요도: medium
• cache.max.bytes.buffering — 모든 스레드에 걸쳐 버퍼링에 사용할 최대 메모리 바이트 수.
  • 유형: long / 기본값: 10485760 / 유효 값: [0,...] / 중요도: medium
• client.id — 내부(메인, 복원, 전역) 컨슈머·프로듀서·admin 클라이언트의 client ID에 쓰는 ID 접두사 문자열. 패턴은 <client.id>-[Global]StreamThread[-<threadSequenceNumber>]-<consumer|producer|restore-consumer|global-consumer>이에요.
  • 유형: string / 기본값: <application.id>-<random-UUID> / 중요도: medium
• default.deserialization.exception.handler — org.apache.kafka.streams.errors.DeserializationExceptionHandler 인터페이스를 구현하는 예외 처리 클래스.
  • 유형: class / 기본값: org.apache.kafka.streams.errors.LogAndFailExceptionHandler / 중요도: medium
• default.key.serde — org.apache.kafka.common.serialization.Serde 인터페이스를 구현하는 키용 기본 직렬화기/역직렬화기 클래스.
  • 유형: class / 기본값: null / 중요도: medium
• default.list.key.serde.inner — org.apache.kafka.common.serialization.Serde 인터페이스를 구현하는 키용 list serde의 기본 내부(inner) 클래스. 이 구성은 default.key.serde가 org.apache.kafka.common.serialization.Serdes.ListSerde로 설정된 경우에만 읽혀요.
  • 유형: class / 기본값: null / 중요도: medium
• default.list.key.serde.type — java.util.List 인터페이스를 구현하는 키용 기본 클래스. 이 구성은 default.key.serde가 org.apache.kafka.common.serialization.Serdes.ListSerde로 설정된 경우에만 읽혀요. list serde 클래스를 쓸 때는 default.list.key.serde.inner를 통해 org.apache.kafka.common.serialization.Serde 인터페이스를 구현하는 inner serde 클래스를 설정해야 해요.
  • 유형: class / 기본값: null / 중요도: medium
• default.list.value.serde.inner — org.apache.kafka.common.serialization.Serde 인터페이스를 구현하는 값용 list serde의 기본 inner 클래스. 이 구성은 default.value.serde가 org.apache.kafka.common.serialization.Serdes.ListSerde로 설정된 경우에만 읽혀요.
  • 유형: class / 기본값: null / 중요도: medium
• default.list.value.serde.type — java.util.List 인터페이스를 구현하는 값용 기본 클래스. 이 구성은 default.value.serde가 org.apache.kafka.common.serialization.Serdes.ListSerde로 설정된 경우에만 읽혀요. list serde 클래스를 쓸 때는 default.list.value.serde.inner를 통해 inner serde 클래스를 설정해야 해요.
  • 유형: class / 기본값: null / 중요도: medium
• default.production.exception.handler — org.apache.kafka.streams.errors.ProductionExceptionHandler 인터페이스를 구현하는 예외 처리 클래스.
  • 유형: class / 기본값: org.apache.kafka.streams.errors.DefaultProductionExceptionHandler / 중요도: medium
• default.timestamp.extractor — org.apache.kafka.streams.processor.TimestampExtractor 인터페이스를 구현하는 기본 타임스탬프 추출기 클래스.
  • 유형: class / 기본값: org.apache.kafka.streams.processor.FailOnInvalidTimestamp / 중요도: medium
• default.value.serde — org.apache.kafka.common.serialization.Serde 인터페이스를 구현하는 값용 기본 직렬화기/역직렬화기 클래스.
  • 유형: class / 기본값: null / 중요도: medium
• deserialization.exception.handler — org.apache.kafka.streams.errors.DeserializationExceptionHandler 인터페이스를 구현하는 예외 처리 클래스.
  • 유형: class / 기본값: org.apache.kafka.streams.errors.LogAndFailExceptionHandler / 중요도: medium
• errors.dead.letter.queue.topic.name — null이 아니면, 오류가 발생했을 때 기본 예외 처리기가 제공된 이름의 토픽에 Dead Letter Queue 레코드를 만들어 보내요. 커스텀 deserialization/production 또는 processing 예외 처리기가 설정되면 이 파라미터는 그 처리기에 대해 무시돼요. 참고: 이 구성은 일반 스트림 처리 작업에만 적용돼요. 전역 상태 저장소 갱신(전역 스레드)에는 적용되지 않아요.
  • 유형: string / 기본값: null / 중요도: medium
• group.protocol — streams가 사용할 그룹 프로토콜. 현재 classic 또는 streams를 지원해요. streams를 지정하면 streams 리밸런스 프로토콜이 사용되고, 그렇지 않으면 classic 그룹 프로토콜이 사용돼요.
  • 유형: string / 기본값: classic / 유효 값: (대소문자 무시) [STREAMS, CLASSIC] / 중요도: medium
• max.task.idle.ms — 조인·병합이 순서가 어긋난 결과를 만들 수 있는지 제어하는 구성. 값은 어떤(전부는 아닌) 입력 파티션을 완전히 따라잡은 스트림 작업이, 프로듀서가 추가 레코드를 보내기를 기다리고 여러 입력 스트림에 걸친 순서가 어긋난 레코드 처리를 피하기 위해 유휴 상태로 머무를 수 있는 최대 시간(밀리초)이에요. 기본값(0)은 프로듀서가 더 많은 레코드를 보내기를 기다리지 않지만, 브로커에 이미 존재하는 데이터를 가져오기는 기다려요. 이 기본값은 브로커에 이미 존재하는 레코드에 대해 Streams가 시간순으로 처리한다는 뜻이에요. -1로 설정하면 유휴를 완전히 비활성화하고 로컬에서 사용 가능한 데이터를 처리하는데, 순서가 어긋난 처리를 만들 수 있어요.
  • 유형: long / 기본값: 0 / 중요도: medium
• max.warmup.replicas — 한 번에 할당할 수 있는 최대 워밍업 복제본 수(설정된 num.standbys를 넘는 추가 standby). 작업을 다른 인스턴스로 재할당했을 때 한 인스턴스에서 워밍업하는 동안 그 작업을 어느 한 인스턴스에서 사용 가능하게 유지하기 위한 목적이에요. 고가용성을 위해 사용되는 추가 브로커 트래픽과 클러스터 상태의 양을 제한하는 데 쓰여요. 최소 1이어야 해요. 참고로 워밍업 복제본 하나는 스트림 작업 하나에 해당해요. 또한 각 워밍업 복제본은 리밸런스 동안(보통 probing.rebalance.interval.ms 구성이 지정하는 빈도로 발생하는 소위 프로빙 리밸런스 동안)에만 활성 작업으로 승격될 수 있어요. 즉 활성 작업이 한 Kafka Streams 인스턴스에서 다른 인스턴스로 마이그레이션될 수 있는 최대 비율은 (max.warmup.replicas / probing.rebalance.interval.ms)로 결정돼요.
  • 유형: int / 기본값: 2 / 유효 값: [1,...] / 중요도: medium
• num.stream.threads — 스트림 처리를 실행할 스레드 수.
  • 유형: int / 기본값: 1 / 중요도: medium
• processing.exception.handler — org.apache.kafka.streams.errors.ProcessingExceptionHandler 인터페이스를 구현하는 예외 처리 클래스. 참고: 이 처리기는 일반 스트림 처리 작업에만 적용돼요. 전역 상태 저장소 갱신(전역 스레드)에는 적용되지 않아요. 전역 스레드에서 발생한 예외는 설정된 uncaught exception handler로 거품처럼 올라가요(bubble up).
  • 유형: class / 기본값: org.apache.kafka.streams.errors.LogAndFailProcessingExceptionHandler / 중요도: medium
• processing.guarantee — 사용할 처리 보증. 가능한 값은 at_least_once(기본값)와 exactly_once_v2(브로커 버전 2.5 이상 필요)예요. exactly-once 처리는 기본적으로 최소 3개 브로커의 클러스터를 요구하며, 이것이 프로덕션 권장 설정이에요. 개발용으로는 브로커 설정 transaction.state.log.replication.factor와 transaction.state.log.min.isr을 조정해 바꿀 수 있어요.
  • 유형: string / 기본값: at_least_once / 유효 값: [at_least_once, exactly_once_v2] / 중요도: medium
• production.exception.handler — org.apache.kafka.streams.errors.ProductionExceptionHandler 인터페이스를 구현하는 예외 처리 클래스.
  • 유형: class / 기본값: org.apache.kafka.streams.errors.DefaultProductionExceptionHandler / 중요도: medium
• replication.factor — 스트림 처리 애플리케이션이 만드는 체인지로그 토픽과 repartition 토픽의 복제 팩터. 기본값 -1(즉 브로커 기본 복제 팩터 사용)은 브로커 버전 2.4 이상이 필요해요.
  • 유형: int / 기본값: -1 / 중요도: medium
• security.protocol — 브로커와 통신할 때 사용하는 프로토콜.
  • 유형: string / 기본값: PLAINTEXT / 유효 값: (대소문자 무시) [SASL_SSL, PLAINTEXT, SSL, SASL_PLAINTEXT] / 중요도: medium
• statestore.cache.max.bytes — 모든 스레드에 걸쳐 상태 저장소 캐시에 사용할 최대 메모리 바이트 수.
  • 유형: long / 기본값: 10485760 (10 mebibytes) / 유효 값: [0,...] / 중요도: medium
• task.assignor.class — org.apache.kafka.streams.processor.assignment.TaskAssignor 인터페이스를 구현하는 작업 할당자 클래스 또는 클래스 이름. 기본은 HighAvailabilityTaskAssignor 클래스예요.
  • 유형: string / 기본값: null / 중요도: medium
• task.timeout.ms — 내부 오류와 재시도로 인해 작업이 멈출 수 있는 최대 시간(밀리초). 그 후 오류가 발생해요. 타임아웃 0ms면 첫 내부 오류에서 작업이 오류를 일으켜요. 0ms보다 큰 타임아웃이면 작업은 오류가 발생하기 전에 최소 한 번 재시도해요.
  • 유형: long / 기본값: 300000 (5분) / 유효 값: [0,...] / 중요도: medium
• topology.optimization — Kafka Streams가 토폴로지를 최적화할지, 어떤 최적화를 적용할지 알려주는 구성. 허용 값은 +NO_OPTIMIZATION+, +OPTIMIZE+, 또는 특정 최적화를 쉼표로 구분한 목록(+REUSE_KTABLE_SOURCE_TOPICS+, +MERGE_REPARTITION_TOPICS+, +SINGLE_STORE_SELF_JOIN+)이에요. 기본은 NO_OPTIMIZATION이에요. 참고: 이 구성을 최적화가 적용되게 하려면 StreamsBuilder#build(Properties)에도 전달해야 해요.
  • 유형: string / 기본값: none / 유효 값: [all, none, reuse.ktable.source.topics, merge.repartition.topics, single.store.self.join] / 중요도: medium
• allow.os.group.write.access — Kafka Streams가 만든 상태 저장소 디렉토리에 OS 그룹이 쓰기 접근할 수 있게 허용할지 여부. 기본은 false.
  • 유형: boolean / 기본값: false / 중요도: low
• application.server — 이 KafkaStreams 인스턴스에서 상태 저장소 발견과 인터랙티브 쿼리(interactive queries)에 사용할 수 있는 사용자 정의 엔드포인트를 가리키는 host:port 쌍.
  • 유형: string / 기본값: "" / 중요도: low
• buffered.records.per.partition — 파티션당 버퍼링할 최대 레코드 수.
  • 유형: int / 기본값: 1000 / 중요도: low
• built.in.metrics.version — 사용할 내장 지표(metrics) 버전.
  • 유형: string / 기본값: latest / 유효 값: [latest] / 중요도: low
• commit.interval.ms — 처리 진행 상황을 커밋하는 빈도(밀리초). at-least-once 처리에서 커밋은 프로세서의 위치(오프셋)를 저장하는 것을 뜻해요. exactly-once 처리에서는 위치 저장과 출력 토픽의 커밋된 데이터가 isolation level read_committed인 컨슈머에게 보이게 만드는 것을 포함한 트랜잭션 커밋을 뜻해요. (참고: processing.guarantee가 exactly_once_v2로 설정되면 기본값은 100, 그렇지 않으면 30000이에요.)
  • 유형: long / 기본값: 30000 (30초) / 유효 값: [0,...] / 중요도: low
• config.providers — ConfigProvider 인터페이스를 구현하는 클래스의 쉼표로 구분된 별칭 이름. 이 설정을 통해 외부 소스(비밀번호, API 키, 기타 자격 증명)에서 구성 데이터를 로드할 수 있어요. 예를 들어 Configuration Providers를 참고해요.
  • 유형: list / 기본값: "" / 중요도: low
• connections.max.idle.ms — 이 구성이 지정하는 밀리초 수가 지난 뒤 유휴 연결을 닫아요.
  • 유형: long / 기본값: 540000 (9분) / 중요도: low
• default.client.supplier — org.apache.kafka.streams.KafkaClientSupplier 인터페이스를 구현하는 클라이언트 공급자 클래스.
  • 유형: class / 기본값: org.apache.kafka.streams.processor.internals.DefaultKafkaClientSupplier / 중요도: low
• default.dsl.store — DSL 연산자가 사용하는 기본 상태 저장소 유형.
  • 유형: string / 기본값: rocksDB / 유효 값: [rocksDB, in_memory] / 중요도: low
• dsl.store.format — DSL 연산자의 상태 저장소 형식을 지정해요. DEFAULT는 상황에 따라 타임스탬프가 있는 또는 일반 상태 저장소를 만들어요. HEADERS는 레코드 헤더를 보존하는 헤더 인식 저장소를 만들어요.
  • 유형: string / 기본값: DEFAULT / 유효 값: (대소문자 무시) [HEADERS, DEFAULT] / 중요도: low
• dsl.store.suppliers.class — DSL 연산자에 끼워 넣을 저장소 구현을 정의해요. org.apache.kafka.streams.state.DslStoreSuppliers 인터페이스를 구현해야 해요.
  • 유형: class / 기본값: org.apache.kafka.streams.state.BuiltInDslStoreSuppliers$RocksDBDslStoreSuppliers / 중요도: low
• enable.metrics.push — (메인, 복원, 전역) 컨슈머·프로듀서·admin 클라이언트의 내부 클라이언트 지표 푸시를 활성화할지 여부. 클러스터에 클라이언트에 해당하는 클라이언트 지표 구독이 있어야 해요.
  • 유형: boolean / 기본값: true / 중요도: low
• log.summary.interval.ms — 요약 정보를 로깅하는 출력 간격(밀리초). 0 이상이면 설정된 시간 간격에 따라 요약 로그가 출력되고, 0보다 작으면 요약 출력이 비활성화돼요.
  • 유형: long / 기본값: 120000 (2분) / 중요도: low
• metadata.max.age.ms — 파티션 리더십 변경을 보지 못했더라도 새 브로커나 파티션을 능동적으로 발견하기 위해 메타데이터 갱신을 강제하는 시간(밀리초).
  • 유형: long / 기본값: 300000 (5분) / 유효 값: [0,...] / 중요도: low
• metadata.recovery.rebootstrap.trigger.ms — metadata.recovery.strategy=rebootstrap로 구성된 클라이언트가 이 간격 동안 마지막으로 알려진 메타데이터의 어떤 브로커에서도 메타데이터를 얻지 못하면, 클라이언트가 bootstrap.servers 구성을 사용해 부트스트랩 과정을 반복해요.
  • 유형: long / 기본값: 300000 (5분) / 유효 값: [0,...] / 중요도: low
• metadata.recovery.strategy — 클라이언트가 아는 브로커 중 아무것도 사용할 수 없을 때 어떻게 복구할지 제어해요. none으로 설정하면 클라이언트가 실패해요. rebootstrap으로 설정하면 클라이언트가 bootstrap.servers를 사용해 부트스트랩 과정을 반복해요. 리부트스트랩은 클라이언트가 브로커와 통신하는 빈도가 너무 낮아서 클라이언트가 메타데이터를 갱신하기 전에 브로커 집합이 완전히 바뀔 수 있을 때 유용해요. 마지막으로 알려진 브로커들이 동시에 사용할 수 없어 보일 때 메타데이터 복구가 트리거돼요. 브로커는 연결이 끊어지고 진행 중인 재시도가 없을 때 사용할 수 없는 것으로 보여요. 클라이언트의 reconnect.backoff.ms와 reconnect.backoff.max.ms를 늘리고 socket.connection.setup.timeout.ms와 socket.connection.setup.timeout.max.ms를 줄이는 것을 고려해요. 또한 metadata.recovery.rebootstrap.trigger.ms 밀리초 동안 어떤 브로커에도 연결을 설정할 수 없거나 서버가 리부트스트랩을 요청하면 리부트스트랩이 트리거돼요.
  • 유형: string / 기본값: rebootstrap / 유효 값: (대소문자 무시) [REBOOTSTRAP, NONE] / 중요도: low
• metric.reporters — 지표 리포터로 사용할 클래스 목록. org.apache.kafka.common.metrics.MetricsReporter 인터페이스를 구현하면 새 지표 생성 시 알림을 받을 클래스를 끼워 넣을 수 있어요. 커스텀 리포터를 설정할 때 org.apache.kafka.common.metrics.JmxReporter가 필요하면 목록에 명시적으로 추가해야 해요.
  • 유형: list / 기본값: org.apache.kafka.common.metrics.JmxReporter / 중요도: low
• metrics.num.samples — 지표를 계산하기 위해 유지하는 샘플 수.
  • 유형: int / 기본값: 2 / 유효 값: [1,...] / 중요도: low
• metrics.recording.level — 지표의 최고 기록 수준. 지표 기록에는 info, debug, trace 세 단계가 있어요. INFO 수준은 시스템 성능과 건강 모니터링에 필요한 필수 지표만 기록해요. 상세를 너무 많이 모으지 않고 핵심 데이터를 수집해 최소 오버헤드를 원하는 프로덕션 환경에 적합해요. DEBUG 수준은 대부분의 지표를 기록해 시스템 동작에 대한 더 상세한 정보를 제공해요. 앱을 디버깅·튜닝하기 위한 더 깊은 통찰이 필요한 개발·테스트 환경에 유용해요. TRACE 수준은 가능한 모든 지표를 기록해 시스템 성능과 동작의 모든 세부 사항을 포착해요. 심층 분석이 필요한 통제된 환경에 가장 좋지만 상당한 오버헤드를 도입할 수 있어요.
  • 유형: string / 기본값: INFO / 유효 값: [INFO, DEBUG, TRACE] / 중요도: low
• metrics.sample.window.ms — 지표 샘플이 계산되는 시간 창.
  • 유형: long / 기본값: 30000 (30초) / 유효 값: [0,...] / 중요도: low
• poll.ms — 입력을 기다리며 막는 시간(밀리초).
  • 유형: long / 기본값: 100 / 중요도: low
• probing.rebalance.interval.ms — 워밍업을 마치고 활성이 될 준비가 된 워밍업 복제본을 프로빙하기 위해 리밸런스를 트리거하기 전에 기다리는 최대 시간(밀리초). 할당이 균형을 이룰 때까지 프로빙 리밸런스가 계속 트리거돼요. 최소 1분이어야 해요.
  • 유형: long / 기본값: 600000 (10분) / 유효 값: [60000,...] / 중요도: low
• processor.wrapper.class — org.apache.kafka.streams.state.ProcessorWrapper 인터페이스를 구현하는 프로세서 래퍼 클래스 또는 클래스 이름. 효과가 있으려면 StreamsBuilder 또는 Topology 생성자에 전달해야 해요.
  • 유형: class / 기본값: org.apache.kafka.streams.processor.internals.NoOpProcessorWrapper / 중요도: low
• rack.aware.assignment.non_overlap_cost — 기존 할당에서 작업을 옮기는 데 드는 비용. 이 구성과 rack.aware.assignment.traffic_cost는 최적화 알고리즘이 크로스 랙 트래픽 최소화를 선호할지, 기존 할당의 작업 이동 최소화를 선호할지를 제어해요. 더 큰 값을 설정하면 org.apache.kafka.streams.processor.internals.assignment.RackAwareTaskAssignor가 기존 할당을 유지하도록 최적화해요. 기본값 null은 각 assignor에서 기본 non_overlap cost 값을 사용한다는 뜻이에요.
  • 유형: int / 기본값: null / 중요도: low
• rack.aware.assignment.strategy — rack aware 할당에 사용하는 전략. Rack aware 할당은 작업을 할당할 때 client.rack과 TopicPartition의 racks를 고려해 크로스 랙 트래픽을 최소화해요. 유효한 설정은: none(기본값, rack aware 할당 비활성화), min_traffic(최소 크로스 랙 트래픽 할당 계산), balance_subtopology(최소 크로스 랙 트래픽 계산 + 서로 다른 클라이언트 간에 같은 서브토폴로지의 작업 균형 맞추기 시도)예요.
  • 유형: string / 기본값: none / 유효 값: [none, min_traffic, balance_subtopology] / 중요도: low
• rack.aware.assignment.tags — standby 복제본을 Kafka Streams 인스턴스 간에 분산하는 데 쓰는 클라이언트 태그 키 목록. 구성하면 Kafka Streams가 각 클라이언트 태그 차원에 걸쳐 standby 작업을 분산하려고 최선을 다해요.
  • 유형: list / 기본값: "" / 유효 값: 최대 5개 요소 포함 목록 / 중요도: low
• rack.aware.assignment.traffic_cost — 크로스 랙 트래픽과 관련된 비용. 이 구성과 rack.aware.assignment.non_overlap_cost는 최적화 알고리즘이 크로스 랙 트래픽 최소화를 선호할지, 기존 할당의 작업 이동 최소화를 선호할지를 제어해요. 더 큰 값을 설정하면 org.apache.kafka.streams.processor.internals.assignment.RackAwareTaskAssignor가 크로스 랙 트래픽 최소화를 위해 최적화해요. 기본값 null은 각 assignor에서 기본 traffic cost 값을 사용한다는 뜻이에요.
  • 유형: int / 기본값: null / 중요도: low
• receive.buffer.bytes — 데이터를 읽을 때 사용하는 TCP 수신 버퍼(SO_RCVBUF) 크기. 값이 -1이면 OS 기본값을 사용해요.
  • 유형: int / 기본값: 32768 (32 kibibytes) / 유효 값: [-1,...] / 중요도: low
• reconnect.backoff.max.ms — 반복적으로 연결 실패한 브로커에 재연결할 때 기다리는 최대 시간(밀리초). 제공되면 호스트별 백오프가 각 연속 연결 실패마다 지수적으로 증가해 이 최대값까지 올라가요. 백오프 증가를 계산한 뒤 연결 폭풍을 피하려고 20%의 무작위 지터가 추가돼요.
  • 유형: long / 기본값: 1000 (1초) / 유효 값: [0,...] / 중요도: low
• reconnect.backoff.ms — 주어진 호스트에 재연결을 시도하기 전에 기다리는 기본 시간. 이렇게 해서 호스트에 빡빡한 루프로 계속 연결하는 걸 피해요. 이 백오프는 클라이언트가 브로커에 하는 모든 연결 시도에 적용돼요. 이 값은 초기 백오프 값이며 각 연속 연결 실패마다 지수적으로 증가해 reconnect.backoff.max.ms 값까지 올라가요.
  • 유형: long / 기본값: 50 / 유효 값: [0,...] / 중요도: low
• repartition.purge.interval.ms — repartition 토픽에서 완전히 소비된 레코드를 삭제하는 빈도(밀리초). 마지막 purge 이후 이 값 이상 지난 뒤에 정리가 발생하지만, 더 늦게 지연될 수도 있어요. (참고: commit.interval.ms와 달리 이 값의 기본값은 processing.guarantee가 exactly_once_v2로 설정돼도 변하지 않아요.)
  • 유형: long / 기본값: 30000 (30초) / 유효 값: [0,...] / 중요도: low
• request.timeout.ms — 클라이언트가 요청의 응답을 기다리는 최대 시간을 제어하는 구성. 타임아웃이 지나기 전에 응답을 받지 못하면 클라이언트는 필요하면 요청을 재전송하고, 재시도가 소진되면 요청을 실패 처리해요.
  • 유형: int / 기본값: 40000 (40초) / 유효 값: [0,...] / 중요도: low
• retry.backoff.ms — 주어진 토픽 파티션에 실패한 요청을 재시도하기 전에 기다리는 시간. 특정 실패 시나리오에서 빡빡한 루프로 요청을 계속 보내는 걸 피해요. 이 값은 초기 백오프 값이며 각 실패 요청마다 지수적으로 증가해 retry.backoff.max.ms 값까지 올라가요.
  • 유형: long / 기본값: 100 / 유효 값: [0,...] / 중요도: low
• rocksdb.config.setter — org.apache.kafka.streams.state.RocksDBConfigSetter 인터페이스를 구현하는 RocksDB config setter 클래스 또는 클래스 이름.
  • 유형: class / 기본값: null / 중요도: low
• send.buffer.bytes — 데이터를 보낼 때 사용하는 TCP 송신 버퍼(SO_SNDBUF) 크기. 값이 -1이면 OS 기본값을 사용해요.
  • 유형: int / 기본값: 131072 (128 kibibytes) / 유효 값: [-1,...] / 중요도: low
• state.cleanup.delay.ms — 파티션이 마이그레이션됐을 때 상태를 삭제하기 전에 기다리는 시간(밀리초). 최소 state.cleanup.delay.ms 동안 수정되지 않은 상태 디렉토리만 제거돼요.
  • 유형: long / 기본값: 600000 (10분) / 중요도: low
• state.cleanup.dir.max.age.ms — 애플리케이션 시작 시 로컬 상태 디렉토리와 체크포인트 파일을 정리하는 시간 기반 임계값. 최소 state.cleanup.dir.max.age.ms 동안 수정되지 않은 상태 디렉토리는 제거돼요.
  • 유형: long / 기본값: -1 / 중요도: low
• upgrade.from — 하위 호환 방식으로 라이브 업그레이드(경우에 따라 다운그레이드 — 업그레이드 가이드 참고)를 허용해요. 기본값은 null이에요. 이 구성을 어떻게·언제 쓸지는 Kafka Streams 업그레이드 가이드를 참고해요. 3.5에서 더 새로운 버전으로 업그레이드할 때는 이 설정을 지정할 필요가 없지만, 2.3 이하에서 이 구성이 있어도 4.0+로 직접 라이브 업그레이드하는 것은 더 이상 지원되지 않아요.
  • 유형: string / 기본값: null / 유효 값: [null, 2.4, 2.5, 2.6, 2.7, 2.8, 3.0, 3.1, 3.2, 3.3, 3.4, 3.5, 3.6, 3.7, 3.8, 3.9, 4.0, 4.1, 4.2] / 중요도: low
• window.size.ms — 윈도우 종료 시간을 계산하기 위해 역직렬화기에 윈도우 크기를 설정해요.
  • 유형: long / 기본값: null / 중요도: low
• windowed.inner.class.serde — 윈도우 레코드의 내부 클래스용 기본 직렬화기/역직렬화기. org.apache.kafka.common.serialization.Serde 인터페이스를 구현해야 해요. 참고: 이 구성을 KafkaStreams 애플리케이션에 설정하면 오류가 발생해요. Plain consumer 클라이언트에서만 쓰려는 것이기 때문이에요.
  • 유형: string / 기본값: null / 중요도: low
• windowstore.changelog.additional.retention.ms — 데이터가 로그에서 너무 일찍 삭제되지 않도록 windows maintainMs에 더하는 시간. 시계 드리프트(clock drift)를 허용하기 위한 거예요. 기본값은 1일이에요.
  • 유형: long / 기본값: 86400000 (1일) / 중요도: low

더 알아보기 (Learn more)

• application.id는 반드시 클러스터 내에서 고유해야 하고, client-ID·group-id·체인지로그 토픽 접두사로 쓰여요.
• processing.guarantee를 exactly_once_v2로 바꾸면 내구성 보증이 강화되지만 브로커 버전·클러스터 구성 요구 사항을 확인해야 해요.
• 상태 저장소는 기본적으로 RocksDB를 쓰며, state.dir이 인스턴스마다 고유해야 해요.