Kafka 구성 튜닝

구성 속성을 지정하는 방법은 배포 유형에 따라 다릅니다. OCP에 Apache Kafka 용 Streams를 배포한 경우 Kafka 리소스를 사용하여 config config 속성을 통해 Kafka 브로커에 대한 구성을 추가할 수 있습니다. RHEL에서 Apache Kafka용 Streams를 사용하면 속성 파일에 구성을 환경 변수로 추가합니다.

사용자 정의 리소스에 구성 속성을 추가할 때 콜론(':')을 사용하여 속성과 값을 매핑합니다.

num.partitions:1

num.partitions:1

속성을 환경 변수로 추가하면 등호('=')를 사용하여 속성과 값을 매핑합니다.

num.partitions=1

num.partitions=1

Kafka 튜닝에 도움이 되는 툴은 다음과 같습니다.

일반적인 브로커 구성에는 주제, 스레드 및 로그와 관련된 속성 설정이 포함됩니다.

...
...

# ...
num.partitions=1
default.replication.factor=3
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
num.network.threads=3
num.io.threads=8
num.recovery.threads.per.data.dir=1
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
group.initial.rebalance.delay.ms=0
zookeeper.connection.timeout.ms=6000
# ...

기본 주제 속성은 주제가 자동으로 생성되는 경우를 포함하여 이러한 속성을 명시적으로 설정하지 않고 생성된 항목에 적용되는 파티션 수와 복제 요소를 설정합니다.

...
...

# ...
num.partitions=1
auto.create.topics.enable=false
default.replication.factor=3
min.insync.replicas=2
replica.fetch.max.bytes=1048576
# ...

고가용성 환경의 경우 복제 요소를 주제의 경우 3 이상으로 늘리고 복제 요인보다 1 미만으로 필요한 최소 in-sync 복제본 수를 설정하는 것이 좋습니다.

auto.create.topics.enable 속성은 아직 존재하지 않는 항목이 생산자와 소비자에 의해 필요할 때 자동으로 생성되도록 기본적으로 활성화됩니다. 자동 주제 생성을 사용하는 경우 num.partitions 를 사용하여 주제의 기본 파티션 수를 설정할 수 있습니다. 그러나 일반적으로 이 속성은 명시적 주제 생성을 통해 주제를 통해 더 많은 제어가 제공되도록 비활성화되어 있습니다.

데이터 지속성 의 경우 주제 구성에 min.insync.replicas 를 설정하고 생산자 구성에서 acks=all 을 사용하여 메시지 전달을 설정해야 합니다.

리더 파티션을 복제하는 각 후속자가 가져온 메시지의 최대 크기(바이트)를 설정하려면 replica.fetch.max.bytes 를 사용합니다. 평균 메시지 크기 및 처리량에 따라 이 값을 변경합니다. 읽기/쓰기 버퍼링에 필요한 총 메모리 할당을 고려할 때 사용 가능한 메모리는 모든 구현자를 곱할 때 최대 복제 메시지 크기를 수용할 수 있어야 합니다.

항목을 삭제할 수 있도록 delete.topic.enable 속성은 기본적으로 활성화됩니다. 프로덕션 환경에서는 실수로 주제 삭제를 방지하기 위해 이 속성을 비활성화하여 데이터가 손실됩니다. 그러나 일시적으로 활성화한 후 주제를 삭제한 다음 다시 비활성화할 수 있습니다.

...
...

# ...
auto.create.topics.enable=false
delete.topic.enable=true
# ...

트랜잭션을 사용하여 생산자의 파티션에 대한 atomic 쓰기를 활성화하는 경우 트랜잭션 상태는 내부 __ Cryostat_state 항목에 저장됩니다. 기본적으로 브로커는 이 항목의 복제 요소 3과 최소 2개의 동기화 복제본으로 구성됩니다. 즉, Kafka 클러스터에 최소 3개의 브로커가 필요합니다.

...
...

# ...
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
# ...

마찬가지로 소비자 상태를 저장하는 내부 __consumer_offsets 주제에는 파티션 수 및 복제 요인에 대한 기본 설정이 있습니다.

...
...

# ...
offsets.topic.num.partitions=50
offsets.topic.replication.factor=3
# ...

프로덕션 환경에서 이러한 설정을 줄이지 마십시오. 프로덕션 환경에서 설정을 늘릴 수 있습니다. 예외적으로 단일broker 테스트 환경에서 설정을 줄일 수 있습니다.

네트워크 스레드는 클라이언트 애플리케이션에서 요청을 생성하고 가져오는 등 Kafka 클러스터에 대한 요청을 처리합니다. 요청 생성은 요청 큐에 배치됩니다. 응답은 응답 큐에 배치됩니다.

리스너당 네트워크 스레드 수는 복제 요소 및 Kafka 클러스터와 상호 작용하는 클라이언트 생산자 및 소비자의 활동 수준을 반영해야 합니다. 요청이 많은 경우 스레드 수를 늘릴 수 있습니다. 시간 스레드의 양을 사용하여 스레드를 추가할 시기를 결정합니다.

혼잡을 줄이고 요청 트래픽을 규제하기 위해 요청 큐에 허용되는 요청 수를 제한할 수 있습니다. 요청 큐가 가득 차면 들어오는 모든 트래픽이 차단됩니다.

I/O 스레드는 요청 대기열에서 요청을 가져와 처리합니다. 스레드를 추가하면 처리량이 향상될 수 있지만 CPU 코어 및 디스크 대역폭의 수는 실용적인 상한을 부과합니다. 최소한 I/O 스레드 수는 스토리지 볼륨 수와 같아야 합니다.

...
...

# ...
num.network.threads=3 

queued.max.requests=500 

num.io.threads=8 

num.recovery.threads.per.data.dir=4 

# ...

모든 브로커의 스레드 풀에 대한 구성 업데이트는 클러스터 수준에서 동적으로 발생할 수 있습니다. 이러한 업데이트는 현재 크기의 절반과 현재 크기의 두 배 사이로 제한됩니다.

디스크 수로 인해 스레드가 느리거나 제한되는 경우 처리량을 개선하기 위해 네트워크 요청의 버퍼 크기를 늘릴 수 있습니다.

...
...

# ...
replica.socket.receive.buffer.bytes=65536
# ...

또한 수신할 수 있는 최대 바이트 수를 늘립니다.

...
...

# ...
socket.request.max.bytes=104857600
# ...

Kafka는 데이터 센터 연결과 같이 Kafka에서 클라이언트로의 대기 시간이 긴 연결을 통해 합리적인 처리량을 달성하기 위해 데이터를 배치합니다. 그러나 대기 시간이 길면 메시지를 보내고 받기 위한 버퍼 크기를 늘릴 수 있습니다.

...
...

# ...
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
# ...

대역폭 지연 제품 계산을 사용하여 버퍼의 최적 크기를 추정할 수 있습니다. 이 계산은 라운드트립 지연 (초 단위)과 링크의 최대 대역폭을 곱하여 최대 처리량을 유지하기 위해 버퍼의 크기를 추정할 수 있습니다.

Kafka는 메시지 데이터를 저장하기 위해 로그를 사용합니다. 로그는 일련의 세그먼트로 구성되며, 여기서 각 세그먼트는 오프셋 기반 및 타임스탬프 기반 인덱스와 연결됩니다. 새 메시지는 활성 세그먼트에 기록되며 이후에 수정되지 않습니다. 소비자의 가져오기 요청을 제공할 때 세그먼트를 읽습니다. 주기적으로 활성 세그먼트는 읽기 전용으로 롤백 되고 이를 대체하기 위해 새 활성 세그먼트가 생성됩니다. 브로커당 주제 파티션당 하나의 활성 세그먼트만 있습니다. 이전 세그먼트는 삭제할 수 있을 때까지 유지됩니다.

브로커 수준의 구성은 로그 세그먼트의 최대 크기(바이트)와 활성 세그먼트가 롤아웃되기 전의 시간(밀리초)을 결정합니다.

...
...

# ...
log.segment.bytes=1073741824
log.roll.ms=604800000
# ...

이러한 설정은 segment.bytes 및 segment.ms 를 사용하여 주제 수준에서 재정의할 수 있습니다. 이러한 값을 낮추거나 올리는 선택은 세그먼트 삭제 정책에 따라 다릅니다. 크기가 클수록 활성 세그먼트에 더 많은 메시지가 포함되어 있으며 덜 자주 롤백됩니다. 또한 세그먼트는 덜 자주 삭제할 수 있습니다.

Kafka에서 로그 정리 정책은 로그 데이터를 관리하는 방법을 결정합니다. 대부분의 경우 삭제 정리 정책을 지정하고 컴팩트 정리 정책에서 사용하는 로그 정리를 활성화하는 클러스터 수준에서 기본 구성을 변경할 필요가 없습니다.

...
...

# ...
log.cleanup.policy=delete
log.cleaner.enable=true
# ...

로그 보존 정책이 적용되면 보존 제한에 도달하면 비활성 로그 세그먼트가 제거됩니다. 이전 세그먼트를 삭제하면 디스크 용량이 초과되는 것을 방지할 수 있습니다.

시간 기반 로그 보존의 경우 시간, 분 또는 밀리초에 따라 보존 기간을 설정합니다.

...
...

# ...
log.retention.ms=1680000
# ...

보존 기간은 메시지가 세그먼트에 추가된 시간을 기반으로 합니다. Kafka는 세그먼트 내에서 최신 메시지의 타임스탬프를 사용하여 해당 세그먼트가 만료되었는지 확인합니다. 밀리초 단위 구성의 우선 순위는 분보다 우선하며, 이는 몇 시간보다 우선 순위가 높습니다. 분 및 밀리초 구성은 기본적으로 null이지만 세 가지 옵션은 유지하려는 데이터에 대한 상당한 수준의 제어를 제공합니다. 동적으로 업데이트할 수 있는 세 가지 속성 중 하나만이므로 기본 설정을 밀리초 구성에 제공해야 합니다.

log.retention.ms 가 -1로 설정된 경우 로그 보존에 시간 제한이 적용되지 않으며 모든 로그가 유지됩니다. 그러나 이 설정은 일반적으로 수정하기 어려운 전체 디스크에 문제가 발생할 수 있으므로 권장되지 않습니다.

크기 기반 로그 보존의 경우 최소 로그 크기(바이트)를 지정합니다.

...
...

# ...
log.retention.bytes=1073741824
# ...

즉 Kafka는 항상 지정된 양의 로그 데이터를 사용할 수 있도록 합니다.

예를 들어 log.retention.bytes 를 1000으로 설정하고 log.segment.bytes 를 300으로 설정하면 Kafka는 4 세그먼트와 활성 세그먼트를 유지하면서 최소 1000바이트를 사용할 수 있습니다. 활성 세그먼트가 가득 차고 새 세그먼트가 생성되면 가장 오래된 세그먼트가 삭제됩니다. 이 시점에서 디스크의 크기는 지정된 1000바이트를 초과할 수 있으며, 잠재적으로 1200 및 1500바이트( 인덱스 파일 제외)를 초과할 수 있습니다.

로그 크기를 사용할 때 발생할 수 있는 문제는 시간 메시지가 세그먼트에 추가되었음을 고려하지 않는다는 것입니다. 정리 정책에 시간 기반 및 크기 기반 로그 보존을 사용하여 필요한 균형을 얻을 수 있습니다. 먼저 도달한 임계값 중 정리를 트리거하는 것은 무엇입니까.

세그먼트 파일이 시스템에서 삭제되기 전에 시간 지연을 추가하려면 모든 주제의 브로커 수준에서 log.segment.delete.delay.ms 를 사용할 수 있습니다.

...
...

# ...
log.segment.delete.delay.ms=60000
# ...

또는 주제 수준에서 file.delete.delay.ms 를 구성합니다.

로그에서 정리를 확인하는 빈도를 밀리초 단위로 설정합니다.

...
...

# ...
log.retention.check.interval.ms=300000
# ...

로그 보존 설정과 관련된 로그 보존 확인 간격을 조정합니다. 보존 크기가 작으면 더 자주 점검해야 할 수 있습니다. 정리 빈도는 디스크 공간을 관리하기에 충분하지만 브로커의 성능에 영향을 미치지 않는 경우가 많습니다.

정리 정책이 로그 압축으로 설정된 경우 로그 헤드 가 표준 Kafka 로그로 작동하며 새 메시지에 대한 쓰기가 순서대로 추가됩니다. 로그 정리기가 작동하는 컴팩트한 로그의 tail 에서 동일한 키가 있는 다른 레코드가 나중에 로그에서 발생하는 경우 레코드가 삭제됩니다. null 값이 있는 메시지도 삭제됩니다. 압축 기능을 사용하려면 Kafka에서 각 키의 최신 메시지가 유지되도록 보장하지만 전체 압축된 로그에 중복이 포함되지 않도록 보장하므로 관련 메시지를 식별하는 키가 있어야 합니다.

그림 3.1. 압축하기 전에 오프셋 위치가 있는 키 값 쓰기를 표시하는 로그

View larger image

Kafka 압축은 메시지를 식별하기 위해 특정 메시지 키에 대해 로그 이름에 존재하는 최신 메시지(가장 높은 오프셋을 가진)를 유지하여 결국 동일한 키가 있는 이전 메시지를 삭제합니다. 최신 상태의 메시지는 항상 사용할 수 있으며 로그 정리기가 실행될 때 특정 메시지의 최신 레코드가 결국 제거됩니다. 메시지를 다시 이전 상태로 복원할 수 있습니다. 레코드는 주변 레코드가 삭제되는 경우에도 원래 오프셋을 유지합니다. 결과적으로 tail은 연속적이지 않은 오프셋을 가질 수 있습니다. tail에서 더 이상 사용할 수 없는 오프셋을 사용하면 다음 상위 오프셋이 있는 레코드가 검색됩니다.

그림 3.2. 압축 후 로그

View larger image

적절한 경우 압축 프로세스에 지연을 추가할 수 있습니다.

...
...

# ...
log.cleaner.delete.retention.ms=86400000
# ...

삭제된 데이터 보존 기간은 데이터가 삭제되기 전에 데이터가 손실되었음을 알리는 시간을 제공합니다.

특정 키와 관련된 모든 메시지를 삭제하려면 생산자가 tombstone 메시지를 보낼 수 있습니다. tombstone에는 null 값이 있으며 해당 키에 대한 해당 메시지가 삭제되었음을 사용자에게 알리는 마커 역할을 합니다. 잠시 후 tombstone 마커만 유지됩니다. 새 메시지가 계속 들어오면, 마커는 log.cleaner.delete.retention.ms 에 지정된 기간 동안 유지되어 사용자가 삭제를 인식할 수 있는 충분한 시간을 허용합니다.

정리할 로그가 없는 경우 대기 상태에 더 많은 시간을 배치하도록 시간(밀리초)을 설정할 수도 있습니다.

...
...

# ...
log.cleaner.backoff.ms=15000
# ...

예를 들어 다음 다이어그램에서는 특정 메시지 키에 대한 최신 메시지(가장 높은 오프셋 포함)만 압축 지점까지 유지됩니다. 보존 지점까지 레코드가 남아 있으면 삭제됩니다. 이 경우 압축 프로세스는 중복을 모두 제거합니다.

그림 3.3. 로그 보존 지점 및 압축 지점

View larger image

Kafka의 주제 로그를 처리하기 위해 컴팩트 정책 및 로그 정리기를 사용하는 경우 메모리 할당을 최적화하는 것이 좋습니다.

deduplication 속성(dedupe.buffer.size)을 사용하여 메모리 할당을 미세 조정할 수 있으며 이는 모든 로그 정리 스레드에서 정리 작업에 할당된 총 메모리를 결정합니다. 또한 buffer.load.factor 속성을 통해 백분율을 정의하여 최대 메모리 사용량 제한을 설정할 수 있습니다.

...
...

# ...
log.cleaner.dedupe.buffer.size=134217728
log.cleaner.io.buffer.load.factor=0.9
# ...

각 로그 항목은 정확히 24바이트를 사용하므로 버퍼가 단일 실행에서 처리할 수 있는 로그 항목 수를 확인하고 그에 따라 설정을 조정할 수 있습니다.

가능한 경우 로그 정리 시간을 줄이려는 경우 로그 정리 스레드 수를 늘리는 것이 좋습니다.

...
...

# ...
log.cleaner.threads=8
# ...

100% 디스크 대역폭 사용량에 문제가 발생하면 로그 정리 I/O를 제한하여 작업을 수행하는 디스크 기능에 따라 읽기/쓰기 작업의 합계가 지정된 이중 값보다 작도록 할 수 있습니다.

...
...

# ...
log.cleaner.io.max.bytes.per.second=1.7976931348623157E308
# ...

일반적으로 명시적 플러시 임계값을 설정하지 않고 운영 체제가 기본 설정을 사용하여 백그라운드 플러시를 수행하도록 하는 것이 좋습니다. 실패한 브로커는 동기화된 복제본에서 복구할 수 있으므로 파티션 복제를 사용하면 단일 디스크에 쓰기보다 데이터 지속성이 향상됩니다.

로그 플러시 속성은 캐시된 메시지 데이터의 주기적인 쓰기를 디스크에 제어합니다. 스케줄러는 로그 캐시의 검사 빈도를 밀리초 단위로 지정합니다.

...
...

# ...
log.flush.scheduler.interval.ms=2000
# ...

메시지가 메모리에 보관되는 최대 시간과 디스크에 쓰기 전에 로그의 최대 메시지 수에 따라 플러시 빈도를 제어할 수 있습니다.

...
...

# ...
log.flush.interval.ms=50000
log.flush.interval.messages=100000
# ...

플러시 간 대기에는 플러시를 수행하기 전에 검사를 수행하는 시간과 지정된 간격이 포함됩니다. 플러시 빈도를 늘리면 처리량에 영향을 미칠 수 있습니다.

애플리케이션 플러시 관리를 사용하는 경우 더 빠른 디스크를 사용하는 경우 플러시 임계값을 낮추는 것이 적합할 수 있습니다.

내결함성을 위해 브로커 간에 파티션을 복제할 수 있습니다. 지정된 파티션의 경우 하나의 브로커가 리더로 선택되고 모든 생성 요청을 처리합니다(로그에 쓰기). 다른 브로커의 파티션 팀은 리더가 실패할 경우 데이터 안정성을 위해 파티션 리더의 파티션 데이터를 복제합니다.

보통은 랙 구성을 통해 Kafka 클러스터가 여러 데이터센터에 걸쳐 있을 때 소비자가 가장 가까운 복제본에서 메시지를 사용할 수 있지만 소비자는 일반적으로 클라이언트에 서비스를 제공하지 않습니다. 차선은 파티션 리더의 메시지를 복제하고 리더가 실패할 경우 복구를 허용하기 위해서만 작동합니다. 복구에는 동기화 내 후속 조치가 필요합니다. 조각화는 리더에게 페치 요청을 보내 동기화 상태를 유지하며, 이 리더에게 메시지를 순서대로 반환합니다. 팔로워는 리더에서 가장 최근에 커밋된 메시지와 함께 배치된 경우 동기화된 것으로 간주됩니다. 리더는 후속자가 요청한 마지막 오프셋을 확인하여 이를 확인합니다. 예기치 않은 리더 선택이 허용되지 않는 한 현재 리더가 실패해야 하는 리더로서 일반적으로 동기화되지 않은 후속 조치를 취할 수 없습니다.

후속 작업이 동기화되지 않은 것으로 간주되기 전에 지연 시간을 조정할 수 있습니다.

...
...

# ...
replica.lag.time.max.ms=30000
# ...

지연 시간이 동기화된 모든 복제본에 메시지를 복제하는 시간과 생산자가 승인을 기다려야 하는 기간을 지정합니다. 후속 프로그램이 가져오기 요청을 작성하고 지정된 지연 시간 내에 최신 메시지를 catch하지 못하면 동기화 내 복제본에서 제거됩니다. 실패한 복제본을 더 빨리 감지할 수 있는 지연 시간을 줄일 수 있지만 이렇게 하면 불필요하게 동기화되지 않는 자국 수를 늘릴 수 있습니다. 올바른 지연 시간 값은 네트워크 대기 시간과 브로커 디스크 대역폭에 따라 다릅니다.

리더 파티션을 더 이상 사용할 수 없는 경우 동기화 중인 복제본 중 하나가 새 리더로 선택됩니다. 파티션의 복제본 목록의 첫 번째 브로커를 선호하는 리더라고 합니다. 기본적으로 Kafka는 리더 배포 주기에 따라 자동 파티션 리더 재조정에 대해 활성화됩니다. 즉 Kafka에서 기본 리더가 현재 리더인지 확인합니다. 리밸런스를 사용하면 브로커와 브로커 간에 리더가 균등하게 분배되지 않습니다.

Apache Kafka용 Cruise Control for Streams를 사용하여 클러스터 전체에서 부하를 균등하게 조정하는 브로커에 대한 복제본 할당을 파악할 수 있습니다. 그 계산은 리더와 팔로워가 경험하는 다양한 부하를 고려합니다. 나머지 브로커는 주요 추가 파티션의 추가 작업을 받기 때문에 실패한 리더는 Kafka 클러스터의 균형에 영향을 미칩니다.

Cruise Control이 실제로 균형을 맞추려면 기본 리더가 파티션을 이끌 필요가 있습니다. Kafka는 기본 리더가 사용되도록 자동으로 확인할 수 있으며 필요한 경우 현재 리더를 변경할 수 있습니다. 이렇게 하면 클러스터가 Cruise Control에서 찾은 균형 있는 상태로 유지됩니다.

리밸런스 검사의 빈도(초)와 리밸런스가 트리거되기 전에 브로커에 허용되는 최대 불균형 비율을 제어할 수 있습니다.

#...
auto.leader.rebalance.enable=true
leader.imbalance.check.interval.seconds=300
leader.imbalance.per.broker.percentage=10
#...

#...
auto.leader.rebalance.enable=true
leader.imbalance.check.interval.seconds=300
leader.imbalance.per.broker.percentage=10
#...

브로커의 백분율 리더 불균형은 브로커가 현재 리더인 현재 파티션 수와 기본 리더인 파티션 수 간의 비율입니다. 백분율을 0으로 설정하여 선호되는 리더가 항상 동기화 중이라고 가정하도록 할 수 있습니다.

리밸런스에 대한 재조정에 더 많은 제어가 필요한 경우 자동 리밸런스를 비활성화할 수 있습니다. 그런 다음 kafka-leader-election.sh 명령줄 도구를 사용하여 리밸런스를 트리거할 시기를 선택할 수 있습니다.

동기화되지 않은 복제본의 리더는 데이터 손실을 보장하므로 정리된 것으로 간주됩니다. 이 작업은 기본적으로 수행됩니다. 그러나 리더십을 위해 동기화되지 않은 복제본이 없으면 어떻게 됩니까? ISR(In-sync) 복제본에는 리더 디스크가 사라진 경우에만 리더가 포함되어 있을 수 있습니다. 최소 동기화 내 복제본 수를 설정하지 않고 하드 드라이브가 실패할 때 파티션 리더와 동기화되는 팔로워가 없는 경우 데이터가 이미 손실됩니다. 그뿐만 아니라 동기화 되지 않은 자국이 없기 때문에 새로운 리더를 선택할 수 없습니다.

Kafka에서 리더 실패를 처리하는 방법을 구성할 수 있습니다.

...
...

# ...
unclean.leader.election.enable=false
# ...

불명확한 리더 선택은 기본적으로 비활성화되어 있습니다. 즉, 동기화되지 않은 복제본이 리더가 될 수 없습니다. 명확한 리더 선택을 통해 이전 리더가 손실되었을 때 다른 브로커가 ISR에 없는 경우 Kafka는 메시지를 쓰거나 읽을 수 있기 전에 해당 리더가 다시 온라인 상태가 될 때까지 기다립니다. 비정형 리더 선택이란 동기화되지 않은 복제본이 리더가 될 수 있지만 메시지를 손실할 위험이 있습니다. 고객의 요구 사항이 가용성 또는 지속성을 선호하는지 여부에 따라 선택할 수 있습니다.

주제 수준에서 특정 주제의 기본 구성을 재정의할 수 있습니다. 데이터 손실 위험을 감수할 수 없는 경우 기본 구성을 그대로 둡니다.

새 소비자 그룹에 가입하는 소비자의 경우 지연을 추가하여 브로커에 불필요한 재조정을 피할 수 있습니다.

...
...

# ...
group.initial.rebalance.delay.ms=3000
# ...

지연은 코디네이터가 멤버가 참여할 때까지 대기하는 시간입니다. 지연이 길어질수록 모든 멤버가 제 시간에 참여하고 재조정을 피할 가능성이 높습니다. 그러나 지연은 기간이 종료될 때까지 그룹이 소비되지 않도록합니다.

모든 소비자에 연결 및 역직렬러 속성이 필요합니다. 일반적으로 추적을 위해 클라이언트 ID를 추가하는 것이 좋습니다.

후속 구성과 관계없이 소비자 구성에서 다음을 수행합니다.

...
...

# ...
bootstrap.servers=localhost:9092 

key.deserializer=org.apache.kafka.common.serialization.StringDeserializer  

value.deserializer=org.apache.kafka.common.serialization.StringDeserializer  

client.id=my-client 

group.id=my-group-id 

# ...

소비자 그룹은 지정된 주제에서 하나 이상의 생산자가 생성한 일반적으로 대규모 데이터 스트림을 공유합니다. 소비자는 group.id 속성을 사용하여 그룹화되므로 메시지를 멤버에 분산할 수 있습니다. 그룹의 소비자 중 한 명이 리더를 선택하고 그룹의 사용자에게 파티션을 할당하는 방법을 결정합니다. 각 파티션은 단일 소비자에만 할당할 수 있습니다.

파티션만큼 많은 소비자가 없는 경우 동일한 group.id 를 사용하여 더 많은 소비자 인스턴스를 추가하여 데이터 소비를 확장할 수 있습니다. 파티션이 있는 것보다 더 많은 소비자를 그룹에 추가하면 처리량은 도움이 되지 않지만 대기 상태의 소비자가 작동을 중지해야 함을 의미합니다. 더 적은 소비자로 처리량 목표를 달성할 수 있는 경우 리소스를 절약할 수 있습니다.

동일한 소비자 그룹 내의 소비자는 오프셋 커밋과 하트비트를 동일한 브로커에 보냅니다. 소비자는 하트비트를 Kafka 브로커로 전송하여 소비자 그룹 내의 활동을 나타냅니다. 따라서 그룹의 소비자 수가 많을수록 브로커에 요청이 로드됩니다.

...
...

# ...
group.id=my-group-id 

# ...

Kafka 주제 파티션이 그룹의 소비자 인스턴스에 배포되는 방식을 결정하는 적절한 파티션 할당 전략을 선택합니다.

파티션 전략은 다음 클래스에서 지원합니다.

partition.assignment.strategy 소비자 구성 속성을 사용하여 클래스를 지정합니다. 범위 할당 전략은 각 사용자에게 파티션 범위를 할당하고 관련 데이터를 함께 처리하려는 경우 유용합니다.

또는 소비자 간 동일한 파티션 배포를 위해 라운드 로빈 할당 전략을 선택합니다. 이는 병렬 처리가 필요한 처리량이 높은 시나리오에 적합합니다.

보다 안정적인 파티션 할당을 위해 고정 및 협업 고정 전략을 고려하십시오. 고정 전략에서는 가능한 경우 리밸런스 중에 할당된 파티션을 유지 관리하는 것을 목표로 합니다. 소비자가 이전에 특정 파티션을 할당한 경우 고정 전략의 우선 순위는 리밸런스 후 동일한 파티션이 있는 동일한 파티션을 유지하는 반면, 실제로 다른 소비자로 이동하는 파티션만 취소 및 재할당합니다. 파티션 할당을 그대로 두면 파티션 조작에 대한 오버헤드가 줄어듭니다. 또한 협업 고정 전략에서는 협업 재조정을 지원하므로 다시 할당되지 않은 파티션에서 중단 없이 소비할 수 있습니다.

사용 가능한 전략이 데이터에 적합하지 않은 경우 특정 요구 사항에 맞게 사용자 정의 전략을 생성할 수 있습니다.

Kafka 브로커는 브로커에 주제, 파티션 및 오프셋 위치 목록에서 메시지를 전송하도록 요청하는 소비자의 가져오기 요청을 받습니다.

소비자는 브로커에 커밋된 것과 동일한 순서로 단일 파티션의 메시지를 관찰합니다. 즉 Kafka는 단일 파티션의 메시지에 대한 순서 만 보장합니다. 반대로, 소비자가 여러 파티션의 메시지를 사용하는 경우 소비자가 관찰한 여러 파티션의 메시지 순서가 전송된 순서를 반영하지 않습니다.

한 주제에서 메시지 순서를 엄격하게 지정하려면 소비자당 하나의 파티션을 사용합니다.

클라이언트 애플리케이션이 KafkaConsumer.poll() 을 호출할 때 반환된 메시지 수를 제어합니다.

fetch.max.wait.ms 및 fetch.min.bytes 속성을 사용하여 Kafka 브로커에서 소비자가 가져온 최소 데이터 양을 늘립니다. 시간 기반 일괄 처리는 fetch.max.wait.ms 를 사용하여 구성되며 크기 기반 일괄 처리는 fetch.min.bytes 를 사용하여 구성됩니다.

소비자 또는 브로커의 CPU 사용률이 높으면 소비자의 요청이 너무 많기 때문일 수 있습니다. 더 적은 요청 및 메시지가 더 큰 배치로 전달되도록 fetch.max.wait.ms 및 fetch.min.bytes 속성을 더 높게 조정할 수 있습니다. 높은 조정을 통해 처리량은 약간의 대기 시간으로 개선됩니다. 또한 생성되는 데이터 양이 낮은 경우 더 높게 조정할 수도 있습니다.

예를 들어 fetch.max.wait.ms 를 500ms로 설정하고 fetch.min.bytes 를 16384바이트로 설정하면 Kafka에서 소비자로부터 가져오기 요청을 수신하면 두 임계값 중 첫 번째에 도달할 때 응답합니다.

반대로 fetch.max.wait.ms 및 fetch.min.bytes 속성을 더 낮게 조정하여 엔드 투 엔드 대기 시간을 개선할 수 있습니다.

...
...

# ...
fetch.max.wait.ms=500 

fetch.min.bytes=16384 

# ...

fetch.max.bytes 속성은 브로커에서 한 번에 가져온 데이터 양에 최대 제한을 바이트 단위로 설정합니다.

max.partition.fetch.bytes 는 각 파티션에 대해 반환되는 데이터의 양에 대한 최대 제한을 바이트 단위로 설정합니다. 이 제한은 항상 브로커에 설정된 바이트 수 또는 max.message.bytes 의 주제 구성보다 커야 합니다.

클라이언트가 사용할 수 있는 최대 메모리 양은 대략적으로 다음과 같이 계산됩니다.

NUMBER-OF-BROKERS * fetch.max.bytes and NUMBER-OF-PARTITIONS * max.partition.fetch.bytes

NUMBER-OF-BROKERS * fetch.max.bytes and NUMBER-OF-PARTITIONS * max.partition.fetch.bytes

메모리 사용량이 이를 수용할 수 있는 경우 이 두 속성의 값을 늘릴 수 있습니다. 각 요청에서 더 많은 데이터를 허용하면 가져오기 요청이 줄어들기 때문에 대기 시간이 향상됩니다.

...
...

# ...
fetch.max.bytes=52428800 

max.partition.fetch.bytes=1048576 

# ...

Kafka 자동 커밋 메커니즘을 사용하면 소비자가 메시지의 오프셋을 자동으로 커밋할 수 있습니다. 활성화하면 소비자는 5000ms 간격으로 브로커 폴링에서 수신된 오프셋을 커밋합니다.

자동 커밋 메커니즘은 편리하지만 데이터 손실 및 복제 위험이 발생합니다. 소비자가 여러 메시지를 가져와서 변환했지만 자동 커밋을 수행할 때 시스템이 소비자 버퍼에서 처리된 메시지와 함께 충돌하면 해당 데이터가 손실됩니다. 메시지를 처리한 후 시스템이 충돌하지만 자동 커밋을 수행하기 전에 시스템의 재조정 후 다른 소비자 인스턴스에서 데이터가 복제됩니다.

자동 커밋은 브로커의 다음 폴링 전에 모든 메시지가 처리되거나 소비자가 닫히기 전에 모든 메시지가 처리되는 경우에만 데이터 손실을 방지할 수 있습니다.

데이터 손실 또는 복제 가능성을 최소화하기 위해 enable.auto.commit 를 false 로 설정하고 클라이언트 애플리케이션을 개발하여 오프셋을 더 많이 제어할 수 있습니다. 또는 auto.commit.interval.ms 를 사용하여 커밋 간 간격을 줄일 수 있습니다.

...
...

# ...
enable.auto.commit=false 

# ...

enable.auto.commit 를 false 로 설정하면 모든 처리가 수행되고 메시지가 사용된 후 오프셋을 커밋할 수 있습니다. 예를 들어 Kafka commitSync 및 commitAsync 커밋 API를 호출하도록 애플리케이션을 설정할 수 있습니다.

commitSync API는 폴링에서 반환된 메시지 배치의 오프셋을 커밋합니다. 일괄 처리의 모든 메시지 처리를 완료하면 API를 호출합니다. commitSync API를 사용하는 경우 배치의 마지막 오프셋이 커밋될 때까지 애플리케이션은 새 메시지를 폴링하지 않습니다. 처리량에 부정적인 영향을 미치는 경우 더 자주 커밋하거나 commitAsync API를 사용할 수 있습니다. commitAsync API는 브로커가 커밋 요청에 응답할 때까지 기다리지 않지만 재조정 시 더 많은 중복을 생성할 위험이 있습니다. 일반적인 접근 방식은 애플리케이션에서 커밋 API를 모두 결합한 후 소비자를 종료하거나 재조정하기 전에 사용된 commitSync API를 결합하여 최종 커밋이 성공했는지 확인하는 것입니다.

...
...

# ...
enable.auto.commit=false
isolation.level=read_committed 

# ...

소비자 그룹 내에서 오류가 발생하는 경우 Kafka는 효과적인 감지 및 복구를 위해 설계된 리밸런스 프로토콜을 제공합니다. 이러한 실패의 잠재적 영향을 최소화하기 위해 하나의 주요 전략은 max.poll.records 속성을 조정하여 시스템 안정성과 효율적인 처리의 균형을 유지하는 것입니다. 이 속성은 소비자가 단일 폴링에서 가져올 수 있는 최대 레코드 수를 결정합니다. fine-tuning max.poll.records 는 제어된 소비률을 유지하는데 도움이 되어 소비자 자신이나 Kafka 브로커를 저해하지 못하도록 합니다.

Kafka는 session.timeout.ms 및 heartbeat.interval.ms 와 같은 고급 구성 속성을 제공합니다. 이러한 설정은 일반적으로 보다 특수한 사용 사례를 위해 예약되어 있으며 표준 시나리오에서 조정이 필요하지 않을 수 있습니다.

session.timeout.ms 속성은 소비자가 소비자 그룹 내에서 활성화되어 있음을 나타내기 위해 Kafka 브로커에 하트비트를 보내지 않고 소비자가 이동할 수 있는 최대 시간을 지정합니다. 소비자가 세션 시간 제한 내에서 하트비트를 전송하지 못하면 비활성 상태로 간주됩니다. 비활성으로 표시된 소비자는 주제의 파티션 재조정을 트리거합니다. session.timeout.ms 속성 값을 너무 낮게 설정하면 false-positive 결과가 발생할 수 있지만 너무 높게 설정하면 실패로 인한 복구가 지연될 수 있습니다.

heartbeat.interval.ms 속성은 소비자가 Kafka 브로커에 하트비트를 보내는 빈도를 결정합니다. 연속 하트비트 간의 더 짧은 간격을 통해 소비자 오류를 더 빠르게 감지할 수 있습니다. 하트비트 간격은 세션 시간 초과보다 일반적으로 3 일까지 작아야 합니다. 하트비트 간격을 줄이면 우발적인 재조정 가능성이 줄어들지만 더 자주 하트비트는 브로커 리소스에 대한 오버헤드를 늘립니다.

auto.offset.reset 속성을 사용하여 오프셋이 커밋되지 않은 경우 소비자가 작동하는 방식을 제어하거나 커밋된 오프셋이 더 이상 유효하지 않거나 삭제되지 않습니다.

소비자 애플리케이션을 처음 배포하고 기존 주제에서 메시지를 읽습니다. group.id 가 처음 사용되므로 __consumer_offsets 항목에 이 애플리케이션에 대한 오프셋 정보가 포함되어 있지 않습니다. 새 애플리케이션은 로그 시작부터 모든 기존 메시지 처리를 시작하거나 새 메시지만 처리할 수 있습니다. 기본 재설정 값은 파티션 끝에 시작되는 latest 이므로 일부 메시지가 누락됨을 의미합니다. 데이터 손실을 피하되 처리 양을 늘리려면 auto.offset.reset 을 가장 빨리 설정하여 파티션 시작 부분에 시작합니다.

또한 브로커에 대해 구성된 오프셋 보존 기간(offsets.retention.minutes)이 종료될 때 메시지가 손실되지 않도록 가장 초기 옵션을 사용하는 것이 좋습니다. 소비자 그룹 또는 독립 실행형 소비자가 비활성 상태이고 보존 기간 동안 오프셋이 없는 경우 이전에 커밋된 오프셋은 __consumer_offsets 에서 삭제됩니다.

...
...

# ...
heartbeat.interval.ms=3000 

session.timeout.ms=45000 

auto.offset.reset=earliest 

# ...

단일 가져오기 요청에서 반환된 데이터 양이 크면 소비자가 처리하기 전에 시간 초과가 발생할 수 있습니다. 이 경우 max.partition.fetch.bytes 를 낮추거나 session.timeout.ms 를 늘릴 수 있습니다.

그룹의 활성 소비자 간 파티션 재조정은 다음을 수행하는 데 걸리는 시간입니다.

리밸런싱 프로세스는 특히 소비자 그룹 클러스터를 롤링 재시작하는 동안 반복적으로 발생하는 경우 서비스의 다운타임을 증가시킬 수 있습니다.

이 경우 그룹 내의 각 소비자 인스턴스에 고유 식별자(group.instance.id)를 할당하여 정적 멤버십 을 도입할 수 있습니다. 정적 멤버십에서는 세션 시간 초과 후 다시 시작하는 동안 소비자 인스턴스를 인식할 수 있도록 지속성을 사용합니다. 결과적으로 소비자는 주제 파티션의 할당을 유지 관리하여 실패하거나 다시 시작한 후 불필요한 재조정을 줄일 수 있습니다.

또한 max.poll.interval.ms 구성을 조정하면 긴 처리 작업으로 인한 재조정을 방지할 수 있으므로 새 메시지에 대한 폴링 간 최대 간격을 지정할 수 있습니다. max.poll.records 속성을 사용하여 각 폴링 중에 소비자 버퍼에서 반환된 레코드 수를 제한합니다. 레코드 수를 줄이면 소비자가 더 적은 메시지를 더 효율적으로 처리할 수 있습니다. 메시지 처리가 불가피한 경우 이러한 작업을 작업자 스레드 풀로 오프로드하는 것이 좋습니다. 이러한 병렬 처리 접근 방식은 대량의 레코드가 있는 소비자의 혼란으로 인한 지연 및 잠재적 재조정을 방지합니다.

...
...

# ...
group.instance.id=UNIQUE-ID 

max.poll.interval.ms=300000 

max.poll.records=500 

# ...

모든 생산자에 연결 및 serializer 속성이 필요합니다. 일반적으로 추적을 위해 클라이언트 ID를 추가하고 생산자에 압축을 사용하여 요청의 배치 크기를 줄이는 것이 좋습니다.

기본 생산자 구성에서 다음을 수행합니다.

...
...

# ...
bootstrap.servers=localhost:9092 

key.serializer=org.apache.kafka.common.serialization.StringSerializer 

value.serializer=org.apache.kafka.common.serialization.StringSerializer 

client.id=my-client 

compression.type=gzip 

# ...

메시지 전달 확인은 메시지가 손실될 가능성을 최소화합니다. 기본적으로 승인은 acks 속성이 acks=all 로 설정된 상태에서 활성화됩니다. 생산자가 브로커의 승인을 기다리고 메시지 전송에 잠재적인 지연을 처리하려면 delivery.timeout.ms 속성을 사용할 수 있습니다.

...
...

# ...
acks=all 

delivery.timeout.ms=120000 

# ...

acks=all 설정은 전달의 가장 강력한 보증을 제공하지만 생산자가 메시지를 전송하고 승인을 수신하는 사이에 대기 시간이 증가합니다. 이러한 강력한 보장이 필요하지 않은 경우 acks=0 또는 acks=1 설정은 전달 보장을 제공하지 않거나 리더 복제본에서 레코드를 로그에 기록한 확인만 제공합니다.

acks=all 에서는 모든 in-sync 복제본이 메시지 전달을 승인할 때까지 기다립니다. 주제의 min.insync.replicas 구성은 필요한 최소 동기화 복제본 승인을 설정합니다. 감사의 수에는 리더와 자국인의 수가 포함됩니다.

일반적인 시작점은 다음 구성을 사용하는 것입니다.

주제를 만들 때 기본 복제 요소를 재정의할 수 있습니다. 주제 구성의 주제 수준에서 min.insync.replicas 를 덮어쓸 수도 있습니다.

Apache Kafka의 스트림은 Kafka의 다중 노드 배포에 대한 구성 파일 예제에서 이 구성을 사용합니다.

다음 표에서는 리더 복제본을 복제하는 팔로워의 가용성에 따라 이 구성이 작동하는 방식을 설명합니다.

3의 주제 복제 요소는 하나의 리더 복제본과 두 명의 유망을 만듭니다. 이 구성에서 단일 팔로워를 사용할 수 없는 경우 프로듀서가 계속될 수 있습니다. 동기화 내 복제본에서 실패한 브로커를 제거하거나 새 리더를 생성하는 동안 일부 지연이 발생할 수 있습니다. 두 번째 팔로워도 사용할 수 없는 경우 메시지 전송이 성공하지 못합니다. 성공적인 메시지 전달을 승인하는 대신 리더는 오류(충분하지 않은 복제본)를 생산자에 보냅니다. 생산자는 동등한 예외를 생성합니다. 재시도 구성을 사용하면 생산자가 실패한 메시지 요청을 다시 보낼 수 있습니다.

멱등 생산자는 메시지가 정확히 한 번 전송되므로 중복을 방지합니다. ID 및 순서 번호는 실패 시에도 전달 순서를 보장하기 위해 메시지에 할당됩니다. 데이터 일관성을 위해 acks=all 을 사용하는 경우 idempotency를 사용하는 것이 좋습니다. 기본적으로 생산자에 대해 멱등이 활성화됩니다. idempotency가 활성화된 경우 메시지 순서를 유지하기 위해 동시 진행 중인 요청 수를 최대 5개로 설정할 수 있습니다.

...
...

# ...
enable.idempotence=true 

max.in.flight.requests.per.connection=5 

acks=all 

retries=2147483647 

# ...

acks=all 을 사용하지 않고 성능 비용 때문에 idempotency를 비활성화하도록 선택하는 경우 순서를 유지하기 위해 in-flight(알림되지 않음) 요청 수를 1로 설정합니다. 그렇지 않으면 Message-B 가 이미 브로커에 기록된 후에만 Message-A 가 성공하지 못하는 상황이 발생할 수 있습니다.

...
...

# ...
enable.idempotence=false 

max.in.flight.requests.per.connection=1 

retries=2147483647
# ...

멱등은 단일 파티션에 정확히 한 번 쓰는 데 유용합니다. 멱등과 함께 사용되는 트랜잭션은 여러 파티션에 정확히 한 번의 쓰기를 허용합니다.

트랜잭션을 사용하면 동일한 트랜잭션 ID를 사용하는 메시지가 한 번 생성되고 모두 해당 로그에 성공적으로 기록되거나 그 중 하나가 표시되지 않습니다.

...
...

# ...
enable.idempotence=true
max.in.flight.requests.per.connection=5
acks=all
retries=2147483647
transactional.id=UNIQUE-ID 

transaction.timeout.ms=900000 

# ...

transactional.id 를 선택하는 것은 트랜잭션 보장이 유지되는 순서대로 중요합니다. 각 트랜잭션 ID는 고유한 주제 파티션 집합에 사용해야 합니다. 예를 들어, 이는 주제 파티션 이름의 외부 매핑을 트랜잭션 ID에 연결하거나 충돌을 방지하는 함수를 사용하여 주제 파티션 이름에서 트랜잭션 ID를 계산하여 수행할 수 있습니다.

일반적으로 시스템의 요구 사항은 지정된 대기 시간 내의 메시지 비율에 대한 특정 처리량 대상을 충족하기 위한 것입니다. 예를 들어 초당 500,000개의 메시지를 대상으로 하는 메시지 중 95%가 2초 이내에 승인됩니다.

생산자의 메시징 의미 체계(메시지 순서 및 지속성)는 애플리케이션의 요구 사항에 따라 정의됩니다. 예를 들어 애플리케이션에서 제공하는 중요한 속성이나 보장을 손상시키지 않고 acks=0 또는 acks=1 을 사용할 수 있는 옵션이 없을 수 있습니다.

브로커 재시작은 높은 백분위 통계에 큰 영향을 미칩니다. 예를 들어 오랜 기간 동안 99번째 백분위 대기 시간이 브로커 재시작에 대한 동작으로 지배됩니다. 이는 벤치마크를 설계하거나 벤치마킹에서 성능 번호를 프로덕션에서 볼 수 있는 성능 수와 비교할 때 고려해야 합니다.

Kafka는 목표에 따라 처리량 및 대기 시간을 위해 생산자 성능 튜닝을 위한 다양한 구성 매개변수와 기술을 제공합니다.

이러한 작업 중에 send() 메서드는 일시적으로 차단됩니다. buffer.memory 가 가득 거나 메타데이터를 사용할 수 없는 경우에도 차단된 상태로 유지됩니다.

배치는 다음 중 하나가 발생할 때까지 큐에 유지됩니다.

latency에 대한 send() 차단의 영향을 최소화하려면 일괄 처리 및 버퍼링 구성을 최적화합니다. linger.ms 및 batch.size 속성을 사용하여 처리량이 높은 단일 생성 요청에 더 많은 메시지를 배치합니다.

...
...

# ...
linger.ms=100 

batch.size=16384 

buffer.memory=33554432 

# ...

...
...

# ...
partitioner.class=my-custom-partitioner 

# ...

Kafka Connect는 소스 외부 데이터 시스템에서 데이터를 가져와서 Kafka Connect 런타임 생산자에 전달하여 대상 클러스터에 복제됩니다.

다음 예제에서는 KafkaConnect 사용자 정의 리소스를 사용한 Kafka Connect에 대한 구성을 보여줍니다.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  replicas: 3
  config:
    offset.flush.timeout.ms: 10000
    # ...
  resources:
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 2Gi
  # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  replicas: 3
  config:
    offset.flush.timeout.ms: 10000
    # ...
  resources:
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 2Gi
  # ...

KafkaConnector 사용자 정의 리소스를 사용하여 관리되는 소스 커넥터에 대한 생산자 구성이 추가됩니다.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: my-source-connector
  labels:
    strimzi.io/cluster: my-connect-cluster
spec:
  class: org.apache.kafka.connect.file.FileStreamSourceConnector
  tasksMax: 2
  config:
    producer.override.batch.size: 327680
    producer.override.linger.ms: 100
    # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: my-source-connector
  labels:
    strimzi.io/cluster: my-connect-cluster
spec:
  class: org.apache.kafka.connect.file.FileStreamSourceConnector
  tasksMax: 2
  config:
    producer.override.batch.size: 327680
    producer.override.linger.ms: 100
    # ...

싱크 커넥터에 대한 소비자 구성이 추가됩니다.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: my-sink-connector
  labels:
    strimzi.io/cluster: my-connect-cluster
spec:
  class: org.apache.kafka.connect.file.FileStreamSinkConnector
  tasksMax: 2
  config:
    consumer.fetch.max.bytes: 52428800
    consumer.max.partition.fetch.bytes: 1048576
    consumer.max.poll.records: 500
    # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: my-sink-connector
  labels:
    strimzi.io/cluster: my-connect-cluster
spec:
  class: org.apache.kafka.connect.file.FileStreamSinkConnector
  tasksMax: 2
  config:
    consumer.fetch.max.bytes: 52428800
    consumer.max.partition.fetch.bytes: 1048576
    consumer.max.poll.records: 500
    # ...

커넥터를 관리하기 위해 KafkaConnector 사용자 정의 리소스 대신 Kafka Connect API를 사용하는 경우 커넥터 구성을 JSON 오브젝트로 추가할 수 있습니다.

curl -X POST \
  http://my-connect-cluster-connect-api:8083/connectors \
  -H 'Content-Type: application/json' \
  -d '{ "name": "my-source-connector",
    "config":
    {
      "connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector",
      "file": "/opt/kafka/LICENSE",
      "topic":"my-topic",
      "tasksMax": "4",
      "type": "source"
      "producer.override.batch.size": 327680
      "producer.override.linger.ms": 100
    }
}'

curl -X POST \
  http://my-connect-cluster-connect-api:8083/connectors \
  -H 'Content-Type: application/json' \
  -d '{ "name": "my-source-connector",
    "config":
    {
      "connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector",
      "file": "/opt/kafka/LICENSE",
      "topic":"my-topic",
      "tasksMax": "4",
      "type": "source"
      "producer.override.batch.size": 327680
      "producer.override.linger.ms": 100
    }
}'

MirrorMaker 2는 소스 클러스터에서 데이터를 가져와서 Kafka Connect 런타임 생산자에 전달하여 대상 클러스터에 복제되도록 합니다.

다음 예제에서는 KafkaMirrorMaker2 사용자 정의 리소스를 사용한 MirrorMaker 2에 대한 구성을 보여줍니다.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker2
spec:
  version: 3.7.0
  replicas: 1
  connectCluster: "my-cluster-target"
  clusters:
  - alias: "my-cluster-source"
    bootstrapServers: my-cluster-source-kafka-bootstrap:9092
  - alias: "my-cluster-target"
    config:
      offset.flush.timeout.ms: 10000
    bootstrapServers: my-cluster-target-kafka-bootstrap:9092
  mirrors:
  - sourceCluster: "my-cluster-source"
    targetCluster: "my-cluster-target"
    sourceConnector:
      tasksMax: 2
      config:
        producer.override.batch.size: 327680
        producer.override.linger.ms: 100
        consumer.fetch.max.bytes: 52428800
        consumer.max.partition.fetch.bytes: 1048576
        consumer.max.poll.records: 500
    # ...
  resources:
    requests:
      cpu: "1"
      memory: Gi
    limits:
      cpu: "2"
      memory: 4Gi

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker2
metadata:
  name: my-mirror-maker2
spec:
  version: 3.7.0
  replicas: 1
  connectCluster: "my-cluster-target"
  clusters:
  - alias: "my-cluster-source"
    bootstrapServers: my-cluster-source-kafka-bootstrap:9092
  - alias: "my-cluster-target"
    config:
      offset.flush.timeout.ms: 10000
    bootstrapServers: my-cluster-target-kafka-bootstrap:9092
  mirrors:
  - sourceCluster: "my-cluster-source"
    targetCluster: "my-cluster-target"
    sourceConnector:
      tasksMax: 2
      config:
        producer.override.batch.size: 327680
        producer.override.linger.ms: 100
        consumer.fetch.max.bytes: 52428800
        consumer.max.partition.fetch.bytes: 1048576
        consumer.max.poll.records: 500
    # ...
  resources:
    requests:
      cpu: "1"
      memory: Gi
    limits:
      cpu: "2"
      memory: 4Gi

Prometheus 및 Grafana를 사용하여 배포를 모니터링하는 경우 MirrorMaker 2 메시지 흐름을 확인할 수 있습니다.

Apache Kafka용 Streams와 함께 제공되는 MirrorMaker 2 Grafana 대시보드 예제에서는 플러시 파이프라인과 관련된 다음 메트릭을 보여줍니다.

이러한 메트릭을 사용하여 메시지 볼륨에 따라 구성을 조정해야 하는지 여부를 평가할 수 있습니다.

대규모 메시지는 시스템 성능에 영향을 미칠 수 있으며 메시지 처리의 복잡성이 발생할 수 있습니다. 방지할 수 없는 경우 구성 옵션을 사용할 수 있습니다. 더 큰 메시지를 효율적으로 처리하고 메시지 흐름에서 블록을 방지하려면 다음 구성을 조정하는 것이 좋습니다.

일괄 요청의 최대 크기가 가장 큰 레코드 배치 크기를 수용하려면 message.max.bytes 만큼 커야 합니다.

message.max.bytes: 10000000
replica.fetch.max.bytes: 10485760

message.max.bytes: 10000000
replica.fetch.max.bytes: 10485760

batch.size: 327680
max.request.size: 10000000

batch.size: 327680
max.request.size: 10000000

fetch.max.bytes: 10000000
max.partition.fetch.bytes: 10485760

fetch.max.bytes: 10000000
max.partition.fetch.bytes: 10485760

더 큰 메시지를 더 효과적으로 처리하기 위해 Kafka Bridge, Kafka Connect 및 MirrorMaker 2와 같은 다른 Kafka 구성 요소에서 사용하는 생산자 및 소비자를 구성할 수도 있습니다.

생산자 구성의 경우 생산자가 생성한 데이터의 배치에 적용되는 Gzip과 같은 compression.type 을 지정합니다. 브로커 구성 compression.type=producer (기본값)를 사용하여 브로커는 생산자가 사용한 압축을 유지합니다. 생산자 및 주제 압축이 일치하지 않을 때마다 브로커는 로그에 추가하기 전에 일괄 처리를 다시 압축해야 브로커 성능에 영향을 미칩니다.

압축은 또한 소비자에 생산자 및 압축 해제 오버헤드에 추가 처리 오버헤드를 추가하지만 배치에 더 많은 데이터를 포함하므로 메시지 데이터가 잘 압축될 때 처리량에 도움이 되는 경우가 많습니다.

생산자 측 압축을 배치 크기의 미세 조정과 결합하여 처리량을 극대화할 수 있습니다. 메트릭을 사용하면 필요한 평균 배치 크기를 측정하는 데 도움이 됩니다.

참조 기반 메시징은 메시지의 규모를 모르는 경우 데이터 복제에 유용합니다. 외부 데이터 저장소는 이 구성이 작동하려면 빠르고 사용 가능한 고가용성이어야 합니다. 데이터는 데이터 저장소에 기록되고 데이터에 대한 참조가 반환됩니다. 생산자는 Kafka에 대한 참조가 포함된 메시지를 보냅니다. 소비자는 메시지에서 참조를 가져와서 이를 사용하여 데이터 저장소에서 데이터를 가져옵니다.

View larger image

메시지 전달에는 더 많은 여행이 필요하므로 엔드 투 엔드 대기 시간이 증가합니다. 이 접근 방식의 또 다른 중요한 단점은 Kafka 메시지가 정리될 때 외부 시스템의 데이터를 자동으로 정리하지 않는다는 것입니다. 하이브리드 접근 방식은 대규모 메시지를 데이터 저장소로만 전송하고 표준 크기의 메시지를 직접 처리하는 것입니다.

생성 클라이언트 애플리케이션은 메시지가 너무 크면 데이터를 직렬화한 다음 청크해야 합니다. 그런 다음 생산자는 Kafka Cryostat ArraySerializer 를 사용하거나 각 청크를 보내기 전에 다시 직렬화하는 것과 유사합니다. 소비자는 전체 메시지가 있을 때까지 메시지 및 버퍼 청크를 추적합니다. 클라이언트 애플리케이션을 사용하는 애플리케이션은 역직렬화 전에 어셈블되는 청크를 수신합니다. 전체 메시지는 청크된 각 메시지 세트에 대한 첫 번째 또는 마지막 청크 오프셋에 따라 소비되는 애플리케이션의 나머지 부분에 전달됩니다.

소비자는 모든 메시지 청크를 수신하고 처리한 후에만 오프셋을 커밋하여 메시지 전달을 정확하게 추적하고 재조정하는 동안 중복을 방지해야 합니다. 청크는 세그먼트에 분산될 수 있습니다. 소비자 측 처리는 나중에 세그먼트를 삭제할 때 청크를 사용할 수 없게 되는 가능성을 포함해야 합니다.

그림 7.1. 인라인 메시징 흐름

View larger image

인라인 메시징은 특히 일련의 대규모 메시지를 병렬로 처리할 때 필요한 버퍼링 때문에 소비자 측에서 성능 오버헤드가 발생합니다. 버퍼에 있는 다른 대규모 메시지의 청크가 불완전하면 메시지의 모든 청크가 사용되었을 때 커밋할 수 없도록 대규모 메시지의 청크가 인터리브될 수 있습니다. 이러한 이유로 메시지 청크를 유지하거나 커밋 논리를 구현하여 버퍼링이 일반적으로 지원됩니다.