7장. Kafka 스트림 구성 속성

type: string
가져오기: high

스트림 처리 애플리케이션의 식별자입니다. Kafka 클러스터 내에서 고유해야 합니다. 1) 기본 클라이언트-id 접두사, 2) 멤버십 관리에 대한 group-id, 3) 변경 로그 주제 접두사로 사용됩니다.

type: list
Importance: high

Kafka 클러스터에 대한 초기 연결을 설정하는 데 사용할 호스트/포트 쌍 목록입니다. 클라이언트는 부트 스트랩을 위해 여기에 지정된 서버에 관계없이 모든 서버를 사용합니다. 이 목록은 전체 서버 세트를 검색하는 데 사용된 초기 호스트에만 영향을 미칩니다. 이 목록은 host1:port1,host2:port2,…​ 형식이어야 합니다. 이러한 서버는 초기 연결이 전체 클러스터 멤버십을 검색하는 데만 사용되므로(이는 동적으로 변경될 수 있음) 전체 서버 세트를 포함할 필요는 없습니다(서버가 다운된 경우).

유형: int
기본값: 0
가져오기: 높음

각 작업에 대한 준비 상태 복제본 수입니다.

type: string
Default: /tmp/kafka-streams
가져오기: high

상태 저장소의 디렉터리 위치입니다. 이 경로는 동일한 기본 파일 시스템을 공유하는 각 스트림 인스턴스에 대해 고유해야 합니다.

type: long
Default: 10000
Valid Values: [0,…​]
가져오기: medium

클라이언트가 활성 작업 할당을 수신할 수 있을 만큼 catch-up으로 간주되는 최대 허용 가능한 지연(수)입니다. 할당 시 처리 전에 나머지 변경 로그를 복원합니다. 리밸런스 중에 처리 일시 정지를 방지하려면 이 구성은 지정된 워크로드의 경우 1분 미만의 복구 시간에 해당합니다. 최소 0이어야 합니다.

type: long
Default: 10485760
Valid Values: [0,…​]
가져오기: 중간

모든 스레드에서 버퍼링에 사용할 최대 메모리 바이트 수입니다.

type: string
Default: ""
가져오기: 중간

'<client.id>-StreamThread-<threadSequenceNumber>-<consumer|producer|restore-consumer>' 패턴과 함께 내부 소비자, 생산자 및 복원 소비자의 클라이언트 ID에 사용되는 ID 접두사 문자열입니다.

class
Default: org.apache.kafka.streams.errors.LogAndFailExceptionHandler
Importance: medium

org.apache.kafka.streams.errors.DeserializationExceptionHandler 인터페이스를 구현하는 예외 처리 클래스입니다.

유형: class
기본값: null
가져오기: 중간

org.apache.kafka.common.serialization.Serde 인터페이스를 구현하는 키의 기본 serializer / deserializer 클래스입니다. serde 클래스를 사용할 때는 org.apache.kafka.common.serialization.Serde 인터페이스를 통해 'default.windowed.key.inner' 또는 'default.windowed.value.serde.inner'를 구현하는 내부 serde 클래스를 설정해야 합니다.

유형: class
기본값: null
가져오기: 중간

org.apache.kafka.common.serialization.Serde 인터페이스를 구현하는 키에 대한 목록 serde의 기본 내부 클래스입니다. default.key.serde 구성이 org.apache.kafka.common.serialization.Serdes.ListSerde 로 설정된 경우에만 이 구성이 읽습니다.

유형: class
기본값: null
가져오기: 중간

java.util.List 인터페이스를 구현하는 키의 기본 클래스입니다. 이 구성은 ' default.key.serde 구성이 org.apache.kafka.common.serialization.Serdes.ListSerde 를 사용할 때 org.apache.kafka.common.serialization.Sde로 설정된 경우에만 읽기되며, 'default.list.kafka.common.serialization .Sde를 통해 org.apache.kafka.common.serialization.Sde 인터페이스를 구현하는 내부 serde 클래스를 설정해야 합니다.

유형: class
기본값: null
가져오기: 중간

org.apache.kafka.common.serialization.Serde 인터페이스를 구현하는 값에 대한 목록 serde의 기본 내부 클래스입니다. 이 구성은 default.value.serde 구성이 org.apache.kafka.common.serialization.Serdes.ListSerde 로 설정된 경우에만 읽을 수 있습니다.

유형: class
기본값: null
가져오기: 중간

java.util.List 인터페이스를 구현하는 값의 기본 클래스입니다. 이 구성은 ' default.value.serde 구성이 org.apache.kafka.common.serialization.Serdes.ListSerde 클래스를 사용할 때 org.apache.kafka.common.serialization.Sde로 설정된 경우에만 읽기되며, 'default.list.kafka.common.serialization .Sde를 통해 org.apache.kafka.common.serialization.Sde 인터페이스를 구현하는 내부 serde 클래스를 설정해야 합니다.

type: class
Default: org.apache.kafka.streams.errors.DefaultProductionExceptionHandler
Importance: medium

org.apache.kafka.streams.errors.ProductionExceptionHandler 인터페이스를 구현하는 예외 처리 클래스입니다.

type: class
Default: org.apache.kafka.streams.processor.FailOnInvalidTimestamp
Importance: medium

org.apache.kafka.streams.processor.Timestamp extractor 인터페이스를 구현하는 기본 타임스탬프 추출기 클래스입니다.

유형: class
기본값: null
가져오기: 중간

org.apache.kafka.common.serialization.Serde 인터페이스를 구현하는 값에 대한 기본 serializer / deserializer 클래스입니다. serde 클래스를 사용할 때는 org.apache.kafka.common.serialization.Serde 인터페이스를 통해 'default.windowed.key.inner' 또는 'default.windowed.value.serde.inner'를 구현하는 내부 serde 클래스를 설정해야 합니다.

유형: long
기본값: 0
가져오기: 중간

이 구성에서는 조인 및 병합이 순서가 아닌 결과를 생성할 수 있는지 여부를 제어합니다. config 값은 일부(전체) 입력 파티션에서 완전히 catch될 때 스트림 작업이 유휴 상태를 유지하여 생산자가 추가 레코드를 보낼 때까지 대기하고 여러 입력 스트림에서 순서가 벗어난 레코드 처리를 방지할 수 있는 최대 시간(밀리초)입니다. 기본(zero)은 생산자가 더 많은 레코드를 보낼 때까지 기다리지 않지만 브로커에 이미 존재하는 데이터를 가져오기 위해 대기합니다. 즉, 브로커에 이미 존재하는 레코드의 경우 Streams는 타임스탬프 순서에 따라 처리합니다. 유휴 상태를 완전히 비활성화하려면 -1로 설정하고 로컬에서 사용 가능한 데이터를 처리하면 순서가 맞지 않는 처리가 발생할 수 있습니다.

type: int
Default: 2
Valid Values: [1,…​]
가져오기: medium

최대 워밍업 복제본 수(구성된 num.standbys 이상으로)는 한 인스턴스에서 작업을 계속 사용할 수 있도록 동시에 할당할 수 있는 반면 다른 인스턴스에서는 다시 할당되었습니다. 고가용성에 사용할 수 있는 추가 브로커 트래픽 및 클러스터 상태를 조정하는 데 사용됩니다. 최소 1 이상이어야 합니다.

유형: int
기본값: 1
가져오기: 중간

스트림 처리를 실행할 스레드 수입니다.

type : string
Default: at_least_once
Valid Values: [at_least_once, exactly_once_once, exactly_once_beta, exactly_once_v2]
Importance: medium

사용되어야 하는 처리 보증입니다. 가능한 값은 at_least_once (기본값) 및 exactly_once_v2 입니다( brokers 버전 2.5 이상 필요). 더 이상 사용되지 않는 옵션은 exactly_once ( brokers 버전 0.11.0 이상) 및 exactly_once_beta (모터 브로커 버전 2.5 이상 필요)입니다. 정확히 한번의 처리에는 기본적으로 프로덕션에 권장되는 설정이 3개 이상의 브로커의 클러스터가 필요합니다. 개발을 위해 broker 설정을 조정하여 이를 변경할 수 있습니다 . transaction.state.log.factor 및 transaction.state.log.min.isr.

type: list
Default: ""
Valid Values: List that contains maximum of 5 elements
Importance: medium

Kafka Streams 인스턴스에 데이터베이스를 배포하는 데 사용되는 클라이언트 태그 키 목록입니다. Kafka 스트림을 구성하면 Kafka Streams가 각 클라이언트 태그 차원에 대해 ReadWriteMany 작업을 배포하는 데 가장 적합한 작업을 수행할 수 있습니다.

유형: int
기본값: -1
가져오기: 중간

스트림 처리 애플리케이션에서 만든 변경 로그 주제 및 다시 파티션 항목의 복제 요인입니다. -1 의 기본값(즉, 브로커 기본 복제 인수 사용)에는 브로커 버전 2.4 이상이 필요합니다.

type: string
Default: PLAINTEXT
Valid Values: [PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL]
가져오기: 중간

브로커와 통신하는 데 사용되는 프로토콜입니다. 유효한 값은 PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL입니다.

type: long
Default: 300000 (5 minutes)
Valid Values: [0,…​]
Importance: medium

작업이 중지되는 최대 시간(밀리초)은 내부 오류로 인해 오류가 발생할 때까지 재시도할 수 있습니다. 시간 초과의 경우 작업에서 첫 번째 내부 오류에 대한 오류가 발생했습니다. 0ms보다 큰 시간 초과의 경우 오류가 발생하기 전에 작업이 한 번 이상 다시 시도합니다.

type: string
Default: none
Valid Values: [none, all]
Importance: medium

구성은 Kafka Streams에 기본적으로 비활성화되어 있는 토폴로지를 최적화해야 하는지 알려줍니다.

유형: string
기본값: ""
가져오기: 낮음

host:port 쌍은 이 KafkaStreams 인스턴스에서 상태 저장소 검색 및 대화형 쿼리에 사용할 수 있는 사용자 정의 끝점을 가리킵니다.

유형: int
기본값: 1000
가져오기: 낮음

파티션당 버퍼에 대한 최대 레코드 수입니다.

type: string
Default: latest
Valid Values: [latest]
가져오기: 낮음

사용할 기본 제공 지표의 버전입니다.

type: long
Default: 30000 (30 seconds)
Valid Values: [0,…​]
가져오기: low

처리 진행 상황을 커밋하는 시간(밀리초)입니다. at-least-once 처리의 경우 커밋은 프로세서의 위치(예: 오프셋)를 저장하는 수단입니다. 정확하게 처리를 수행하려면 위치를 저장하고 격리 수준이 read_committed인 사용자에게 출력 항목에 커밋된 데이터를 만드는 작업이 포함된 트랜잭션을 커밋합니다.For exactly-once processing, it means to commit the transaction which includes to save the position and to make the committed data in the output topic visible to consumers with isolation level read_committed. (참고로는 processing.guarantee 가 exactly_once_v2,exactly_once, 기본값은 100 입니다. 그렇지 않으면 기본값은 30000 입니다.

유형: long
기본값: 540000 (9분)
가져오기: 낮음

이 구성에서 지정한 시간(밀리초) 후 유휴 연결을 종료합니다.

type: string
Default: concksDB
Valid Values: [rocksDB, in_memory]
가져오기: 낮음

DSL Operator에서 사용하는 기본 상태 저장소 유형입니다.

type: long
Default: 300000 (5 minutes)
Valid Values: [0,…​]
가져오기: low

이 기간(밀리초) 후 새 브로커 또는 파티션을 사전에 검색하기 위해 파티션 리더십 변경을 보지 못하더라도 메타데이터를 강제로 새로 고침합니다.

type: list
Default: ""
가져오기: 낮음

메트릭 보고자로 사용할 클래스 목록입니다. org.apache.kafka.common.metrics.MetricsReporter 인터페이스를 구현하면 새 지표 생성에 대한 알림을 받을 클래스를 연결할 수 있습니다. JmxReporter는 항상 통계를 등록하기 위해 포함됩니다.

type: int
Default: 2
Valid Values: [1,…​]
가져오기: 낮음

메트릭으로 유지 관리되는 샘플 수입니다.

type: string
Default: INFO
Valid Values: [INFO, DEBUG, TRACE]
가져오기: low

메트릭에 대해 가장 높은 기록 수준입니다.

type: long
Default: 30000 (30 seconds)
Valid Values: [0,…​]
가져오기: low

지표 샘플이 계산되는 시간 창입니다.

유형: long
기본값: 100
가져오기: 낮은

입력 대기를 차단하는 시간(밀리초)입니다.

type: long
Default: 600000 (10 minutes)
Valid Values: [60000,…​]
가져오기: 낮음

재조정을 트리거하기 전에 대기하는 최대 시간(밀리초)은 온난화를 완료하고 활성 상태가 될 준비가 된 워밍업 복제본을 검색합니다. 프로빙 리밸런스는 할당이 균형을 유지할 때까지 계속 트리거됩니다. 최소 1분 이상이어야 합니다.

type: int
기본값: 32768(32 kibibytes)
유효 값: [-1,…​]
가져오기: 낮음

데이터를 읽을 때 사용할 TCP 수신 버퍼(SO_RCVBUF)의 크기입니다. 값이 -1이면 OS 기본값이 사용됩니다.

type: long
Default: 1000 (1 second)
Valid Values: [0,…​]
가져오기: low

브로커에 다시 연결할 때 대기하는 최대 시간(밀리초)입니다. 제공되는 경우 호스트당 백오프는 연속된 연결 실패마다 최대 최대 기하급수적으로 증가합니다. 백오프 증가를 계산한 후 연결 문제가 발생하지 않도록 20 % 임의 지터가 추가됩니다.

type: long
Default: 50
Valid Values: [0,…​]
가져오기: low

지정된 호스트에 다시 연결하기 전에 대기하는 기본 시간입니다. 이렇게 하면 엄격한 루프에서 호스트에 반복적으로 연결하는 것을 방지할 수 있습니다. 이 백오프는 클라이언트가 브로커에 대한 모든 연결 시도에 적용됩니다.

type: long
Default: 30000 (30 seconds)
Valid Values: [0,…​]
가져오기: low

다시 분할 주제에서 완전히 사용된 레코드를 삭제할 수 있는 시간(밀리초)입니다. 제거는 마지막 제거 이후 적어도 이 값 후에 발생하지만 나중에 지연될 수 있습니다. (참고로, commit.interval.ms 와 달리, processing.guarantee 가 exactly_once_v2로 설정되어 있을 때 이 값의 기본값은 변경되지 않은 상태로 유지됩니다.

type: int
Default: 40000 (40 seconds)
Valid Values: [0,…​]
가져오기: low

구성은 클라이언트가 요청 응답을 기다리는 최대 시간을 제어합니다. 시간 초과가 경과하기 전에 응답이 수신되지 않으면 필요한 경우 클라이언트가 요청을 다시 전송하거나 재시도가 고갈되면 요청이 실패합니다.

type: int
Default: 0
Valid Values: [0,…​,2147483647]
가져오기: low

값을 0보다 크게 설정하면 클라이언트가 잠재적으로 일시적인 오류와 함께 실패하는 모든 요청을 다시 보냅니다. 이 값을 0 또는 MAX_VALUE 로 설정하고 해당 시간 초과 매개변수를 사용하여 클라이언트가 요청을 다시 시도해야 하는 기간을 제어하는 것이 좋습니다.

type: long
Default: 100
Valid Values: [0,…​]
가져오기: low

지정된 주제 파티션에 실패한 요청을 재시도하기 전에 대기하는 시간입니다. 이렇게 하면 일부 실패 시나리오에서 엄격한 루프에서 반복적으로 요청을 보내는 것을 방지할 수 있습니다.

유형: class
기본값: null
가져오기: 낮음

org.apache.kafka.streams.state.RocksDBConfigSetter 인터페이스를 구현하는 CloudEvents DB 구성 setter 클래스 또는 클래스 이름입니다.

type: int
기본값: 131072(128 키비바이트)
유효 값: [-1,…​]
가져오기: 낮음

데이터를 전송할 때 사용할 TCP 전송 버퍼(SO_SNDBUF)의 크기입니다. 값이 -1이면 OS 기본값이 사용됩니다.

유형: long
기본값: 600000(10분)
가져오기: 낮음

파티션이 마이그레이션될 때 상태를 삭제하기 전에 대기하는 시간(밀리초)입니다. 최소 state.cleanup.delay.ms 용으로 수정되지 않은 디렉터리만 제거됩니다.

type : string
기본값
기본값: [null, 0.10.0, 0.10.1, 0.10.2, 0.11.0, 1.0, 1.1, 2.0, 2.1, 2.2, 2.3, 2.4, 2.5, 2.6, 2.7, 2.8, 3.0, 3.1, 3.2]
가져오기: 낮은

이전 버전과 호환되는 방식으로 업그레이드할 수 있습니다. 이는 [0.10.0, 1.1]에서 2.0+로 업그레이드하거나 [2.0, 2.3]에서 2.4+로 업그레이드할 때 필요합니다. 3.3에서 최신 버전으로 업그레이드하는 경우 이 구성을 지정할 필요가 없습니다. 기본값은 null 입니다. 허용되는 값은 "0.10.0", "0.10.2", "0.10.2", "0.11.0", "1.1", "1.1", "2.1", "2.2", "2.3", "2.4", "2.5", "2.6", "2.7", "2.8", "3.0", "3.0", "3.1", "3", "3"입니다.

유형: long
기본값: null
가져오기: 낮음

창 끝 시간을 계산하기 위해 deserializer의 창 크기를 설정합니다.

유형: string
기본값: null
가져오기: 낮음

창된 레코드의 내부 클래스에 대한 기본 serializer / deserializer입니다. org.apache.kafka.common.serialization.Serde 인터페이스를 구현해야 합니다. KafkaStreams 애플리케이션에서 이 구성을 설정하면 Plain Consumer 클라이언트에서만 사용할 수 있으므로 오류가 발생합니다.

유형: long
기본값: 86400000 (1일)
가져오기: 낮음

창에 추가되어 데이터가 조기 로그에서 삭제되지 않도록 합니다. 클럭 드리프트를 허용합니다. 기본값은 1일입니다.