Apache Kafka에 대해서 멋있는 특성을 모르실 수도 있어요.

아파치 카프카는 광범위하게 분포된 이벤트 스트리밍 플랫폼이다.본고에서 저는 ApacheKafka의 기본 원리, 예를 들어 주제, 구역, 편이량 또는 동기화 복사본이 무엇인지 설명하지 않을 것입니다.ApacheKafka의 작동 방식과 기본 원리를 이해할 수 있는 리소스를 많이 찾을 수 있습니다.나는 당신에게 들어보지 못했을 수도 있는 기능들을 소개할 것입니다.

미터기자.

ApacheKafka 클라이언트: 생산자와 소비자는 JMX 지표를 사용하여 모니터링할 수 있습니다.소비자here와 생산자here에 공개된 모든 JMX 지표를 조회할 수 있습니다.
기본적으로 Apache Kafka 클라이언트 사용 클래스JmxReporter는 JMX 메트릭을 공개합니다.이러한 구현 인터페이스MetricsReporter

public interface MetricsReporter extends Reconfigurable, AutoCloseable {
    void init(List<KafkaMetric> var1);

    void metricChange(KafkaMetric var1);

    void metricRemoval(KafkaMetric var1);

    void close();

    default Set<String> reconfigurableConfigs() {
        return Collections.emptySet();
    }

    default void validateReconfiguration(Map<String, ?> configs) throws ConfigException {
    }

    default void reconfigure(Map<String, ?> configs) {
    }

    @Evolving
    default void contextChange(MetricsContext metricsContext) {
    }
}

따라서 이 인터페이스를 실현하고 설정 metric.reporters 을 설정하여 카프카 지표를 공개하는 방법을 정의할 수 있습니다

settings.put(ConsumerConfig.METRIC_REPORTER_CLASSES_CONFIG, List.of(MyMetricReporter.class));

클라이언트 차단기

클라이언트 차단기는 수신하거나 생성한 기록을 캡처하고 변이할 수 있습니다.모든 ApacheKafka 클라이언트: 생산자와 소비자에 대해 차단기 종류를 정의할 수 있습니다.
생산자에 대해 당신은 반드시 인터페이스를 실현해야 합니다ProducerInterceptor

public interface ProducerInterceptor<K, V> extends Configurable {
    ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);

    void onAcknowledgement(RecordMetadata var1, Exception exception);

    void close();
}

소비자에게 인터페이스ConsumerInterceptor

public interface ConsumerInterceptor<K, V> extends Configurable, AutoCloseable {
    ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records);

    void onCommit(Map<TopicPartition, OffsetAndMetadata> partitions);

    void close();
}

이 인터페이스가 실현되면 설정interceptor.classes을 사용하여 클라이언트에 차단기를 추가합니다.

settings.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,
        List.of(MyConsumerInterceptor.class));

마이크로서비스 아키텍처에서 Apache Kafka를 사용하고 추적 감독을 추가하고자 하는 경우 특히 유용합니다.예를 들어 OpenTracing을 사용하면 볼 수 있다opentracing-contrib/java-kafka-client. 이것은 TracingProducerInterceptor와 TracingConsumerInterceptor를 제공한다.이 차단기들은 기록 헤더에서 상하문에 걸쳐 주입되고 검색된다.

파티션 할당 및 공통 파티션

10개의 테마를 가지고 있고, 테마마다 구역이 하나씩 있다면, 10개의 소비자 실례를 포함하는 소비자 그룹을 만들 수 있습니다.소비자들은 각 구역을 어떻게 분배할 것인가?너는 소비자가 하나의 구역을 소비한다고 가정할 수 있다.실제로 이것은 사용자의 실례 사이에서 구역을 분배하는 전략에 달려 있다.
기본적으로 소비자는 사용 RangeAssignor 으로 구성됩니다.이 전략의 목표는 모든 구독 주제의 구역을 같은 소비자로 공동으로 현지화하는 것이다.따라서 첫 번째 질문에 대답하기 위해 한 사용자는 모든 구역에 분배되고 나머지 사용자는 빈 상태가 된다.

기본적으로 사용되는 이 정책 설계는 공통 구역 테마에 사용됩니다.공통 구역 테마는 같은 수량의 구역을 가진 테마로 같은 구역 구분기와 같은 구역 키를 사용하여 메시지를 생성합니다.
단, 소비자 그룹에서 사용하는 소비자 수를 최대한 늘리는 것이 목표라면 속성partition.assignment.strategy을 사용하여 기본 분배 정책을 사용 RoundRobinAssignor 으로 변경해야 합니다.

랙 의식

클러스터의 각 에이전트에 속성broker.rack을 지정하면 Apache Kafka는 지정된 모든 랙에 동일한 파티션의 복제본을 전파합니다.이 기능은 랙 가동 중지 시 데이터 손실의 위험을 제한합니다.랙은 클라우드 공급업체 영역 또는 데이터 센터 위치일 수 있습니다.

그러나 Apache Kafka 클라이언트는 항상 조닝(Zoning) 책임자와 상호 작용하므로 에이전트와 다른 영역에 있는 클라이언트의 경우 문제가 발생할 수 있습니다.

이 그림에서 소비자는 B구역에 위치하고 구역 담당자는 A구역에 위치합니다. 구역별 요청 수량을 증가시키고 일부 클라우드 공급자의 비용 계산을 증가시킬 수 있습니다.
이를 피하기 위해 아파치카프카 2.4는 추종자 캡처 개념을 도입했다.이는 사용자가 구역 지도자가 아닌 수행자로부터 데이터를 직접 얻을 수 있다는 뜻이다.이 기능을 사용하려면 에이전트가 속성broker.rack을 원하는 위치로 설정하고 속성replica.selector.class을 org.apache.kafka.common.replica.RackAwareReplicaSelector로 설정해야 합니다.다른 한편, 소비자는 반드시 설정client.rack을 해서 그들이 어느 위치에 있는지 판단해야 한다.

카프카

Kafka Lag Exporter 아파치 카프카 프로젝트의 일부도 아니고 융합 플랫폼도 아니다.이는 프로메테우스(Prometheus), 그래파이트(Graphite) 또는 XDB 등 기자들에게 소비자 대기 정보를 출력하기 위한 아파치-2.0 허가를 받은 오픈소스 프로젝트다.Kafka Lag Exporter는 소비자의 대기 상태를 모니터링할 수 있지만, 대기 시간을 가늠할 수도 있다.이 지표는 한 소비층이 지난번에 세운 기록보다 시간적으로 뒤떨어지는 정도를 보여 준다.그것은 소비자 응용 프로그램의 실제 지연을 보여 준다.이 추정에 대한 자세한 내용은 설명서의 장을 참조하십시오Estimate Consumer Group Time Lag.

클러스터 링크

다중 데이터 센터 아키텍처에서 클러스터 링크는 서로 다른 두 Apache Kafka 클러스터 간에 링크를 만드는 기능입니다.링크가 설정되면 테마를 거울로 볼 수 있습니다.
거울 테마는 원본 그룹의 특정한 테마와 설정된 모든 데이터를 대상 그룹의 같은 이름의 테마로 복사하는 것을 의미합니다.

Unlike, Replicator and MirrorMaker2, Cluster Linking does not require running Connect to move messages from one cluster to another, ensuring that the offsets are preserved from one cluster to another. We call this “byte-for-byte” replication. Whatever is on the source, will be mirrored precisely on the destination cluster.
https://docs.confluent.io/platform/current/multi-dc-deployments/cluster-linking/index.html

대상 테마는 '바이트 단위' 복사를 실행합니다. 한 그룹에서 다른 그룹으로 이동하는 편이도를 보존하고, 원본 테마 설정도 복제합니다. 예를 들어 보존, 원본 테마의 구역 수를 복제합니다.
집단 링크는 매우 유망한 기능이지만, 지금은 미리 보기 기능이다.

A preview feature is a component of the Confluent Platform that is being introduced to gain early feedback from developers. This feature can be used for evaluation and non-production testing purposes or to provide feedback to Confluent.

Confluent는 이 기능을 사용하기 위해 a demonstration project를 제공합니다.

계층형 스토리지

계층형 스토리지는 에이전트가 아닌 외부 스토리지에 데이터를 저장하는 로컬 디스크의 기능입니다.그 생각은 데이터 저장과 데이터 처리를 분리하는 것이다.데이터를 수개월, 수년 또는 무기한으로 보존할 수 있습니다.
Apache Kafka 클러스터는 AWS S3, GCP GCS, 순수 스토리지 FlashBlade 등 세 가지 지원되는 계층형 스토리지 중 하나로 구성할 수 있습니다.
예: Amazon S3:

confluent.tier.feature=true
confluent.tier.enable=true
confluent.tier.backend=S3
confluent.tier.s3.bucket=<BUCKET_NAME>
confluent.tier.s3.region=<REGION>

그런 다음 계층형 스토리지가 포함된 주제를 만들 수 있습니다.

kafka-topics --bootstrap-server localhost:9092   \
  --create --topic tiered-storage-topic \
  --partitions 6 \
  --replication-factor 3 \
  --config confluent.tier.enable=true \
  --config confluent.tier.local.hotset.ms=3600000 \
  --config retention.ms=604800000

retention.ms의 작업 원리는 일반 주제와 같다.로컬 또는 계층형 스토리지에 데이터를 저장할 시간을 정의합니다.confluent.tier.local.hotset.ms 비활성 세그먼트가 로컬 스토리지에 유지되는 시간(ms)을 정의합니다.세그먼트가 만료되면 로컬 스토리지에서 제거되어 계층형 스토리지에 저장됩니다.

도구책

https://docs.confluent.io/platform/current/kafka/post-deployment.html#balancing-replicas-across-racks

https://docs.confluent.io/platform/current/multi-dc-deployments/cluster-linking/index.html

https://docs.confluent.io/platform/current/multi-dc-deployments/multi-region.html#follower-fetching

https://github.com/lightbend/kafka-lag-exporter#estimate-consumer-group-time-lag

https://docs.confluent.io/platform/current/multi-dc-deployments/cluster-linking/tutorial.html

https://docs.confluent.io/platform/current/kafka/tiered-storage.html