OpenShift 개요에서 Apache Kafka용 스트림

Kafka의 기본 데이터 스트림 처리 기능 및 구성 요소 아키텍처는 다음을 제공할 수 있습니다.

Kafka의 기능은 다음에 적합합니다.

AMQ Streams는 OpenShift에서 Kafka를 실행하기 위한 컨테이너 이미지 및 운영자를 제공합니다. AMQ Streams Operator는 OpenShift에서 Kafka를 효과적으로 관리하기 위해 전문 지식으로 구축된 것입니다.

Operator는 다음 프로세스를 단순화합니다.

Kafka 클러스터는 메시지 전송을 담당하는 브로커로 구성됩니다.

Zookeeper는 클러스터 관리에 사용됩니다. KRaft (Kafka Raft metadata) 모드에서 Kafka를 배포할 때 Kafka 노드 내에 브로커 및 컨트롤러 역할을 통합하여 클러스터 관리가 간소화되어 Zoo Cryostat의 필요성이 제거됩니다. Kafka 노드는 브로커, 컨트롤러 또는 둘 다의 역할을 수행합니다. 역할은 노드 풀을 사용하여 AMQ Streams에 구성됩니다.

다른 각 Kafka 구성 요소는 Kafka 클러스터와 상호 작용하여 특정 역할을 수행합니다.

AMQ Streams 사용 환경을 극대화하려면 Kafka가 메시지 브로커로 작동하는 방식을 이해해야 합니다.

생산자와 소비자는 브로커를 통해 메시지를 보내고 수신(게시 및 구독)합니다. 메시지는 선택적 키와 메시지 데이터가 포함된 값 과 헤더 및 관련 메타데이터로 구성됩니다. 키는 메시지의 제목 또는 메시지의 속성을 식별하는 데 사용됩니다. 메시지는 일괄 처리로 전달되며, 일괄 처리 및 레코드에는 레코드의 타임스탬프 및 오프셋 위치와 같이 클라이언트별로 필터링 및 라우팅하는 데 유용한 세부 정보를 제공하는 헤더 및 메타데이터가 포함되어 있습니다.

Kafka Connect는 커넥터 인스턴스를 사용하여 다른 시스템과 통합하여 데이터를 스트리밍합니다.

Kafka Connect는 작업자 Pod에 걸쳐 데이터 스트리밍 작업 및 커넥터 구성을 시작할 때 기존 커넥터 인스턴스를 로드하고 배포합니다. 작업자는 커넥터 인스턴스에 대한 작업을 실행합니다. 각 작업자는 Kafka Connect 클러스터의 내결함성을 높이기 위해 별도의 Pod로 실행됩니다. 작업자보다 많은 작업이 있는 경우 작업자에 여러 작업이 할당됩니다. 작업자가 실패하면 Kafka Connect 클러스터의 활성 작업자에 작업이 자동으로 할당됩니다.

스트리밍 데이터에 사용되는 주요 Kafka Connect 구성 요소는 다음과 같습니다.

Kafka 브리지는 다음과 같은 작업을 수행하는 방법으로 Kafka 클러스터에 대한 HTTP 요청을 지원합니다.

이 메서드는 JSON 응답 및 HTTP 응답 코드 오류 처리를 제공합니다. 메시지는 JSON 또는 바이너리 형식으로 보낼 수 있습니다.

Kafka 브리지를 사용하여 내부 및 외부 HTTP 클라이언트 애플리케이션을 Kafka 클러스터와 통합할 수 있습니다.

AMQ Streams는 Cluster Operator를 사용하여 클러스터를 배포 및 관리합니다. 기본적으로 AMQ Streams를 배포할 때 단일 Cluster Operator 복제본이 배포됩니다. 리더 선택 항목이 있는 복제본을 추가하여 중단 시 추가 Cluster Operator가 대기 상태가 되도록 할 수 있습니다.

Cluster Operator는 다음 Kafka 구성 요소의 클러스터를 관리합니다.

클러스터는 사용자 정의 리소스를 사용하여 배포됩니다.

예를 들어 Kafka 클러스터를 배포하려면 다음을 수행합니다.

Cluster Operator는 Kafka 리소스의 구성을 통해 다음 AMQ Streams Operator도 배포할 수 있습니다.

배포 시 Entity Operator 내에서 Topic Operator 및 User Operator가 작동합니다.

AMQ Streams Drain cleaner 배포와 함께 Cluster Operator를 사용하여 Pod 제거를 지원할 수 있습니다. AMQ Streams Drain cleaner를 배포하면 Cluster Operator를 사용하여 OpenShift 대신 Kafka Pod를 이동할 수 있습니다. AMQ Streams DrainCleaner는 롤링 업데이트 주석으로 제거되는 Pod에 주석을 추가합니다. 이 주석은 Cluster Operator에 롤링 업데이트를 수행하도록 지시합니다.

Topic Operator는 KafkaTopic 리소스를 통해 Kafka 클러스터에서 주제를 관리하는 방법을 제공합니다.

Topic Operator는 Kafka 주제를 설명하는 KafkaTopic 리소스를 조사하고 Kafka 클러스터에서 올바르게 구성되었는지 확인하여 Kafka 주제를 관리합니다.

KafkaTopic 이 생성, 삭제 또는 변경되면 Topic Operator는 Kafka 주제에서 해당 작업을 수행합니다.

KafkaTopic 을 애플리케이션 배포의 일부로 선언하고 Topic Operator는 Kafka 주제를 관리합니다.

User Operator는 KafkaUser 리소스를 통해 Kafka 클러스터에서 사용자를 관리하는 방법을 제공합니다.

User Operator는 Kafka 사용자를 설명하는 KafkaUser 리소스를 조사하고 Kafka 클러스터에서 올바르게 구성되었는지 확인하여 Kafka 클러스터의 Kafka 사용자를 관리합니다.

KafkaUser 가 생성, 삭제 또는 변경되면 User Operator가 Kafka 사용자에게 해당 작업을 수행합니다.

KafkaUser 리소스를 애플리케이션 배포의 일부로 선언하고 User Operator는 Kafka 사용자를 관리합니다. 사용자의 인증 및 권한 부여 메커니즘을 지정할 수 있습니다. 예를 들어 사용자가 브로커에 대한 액세스를 단조하지 않도록 Kafka 리소스의 사용을 제어하는 사용자 할당량 을 구성할 수도 있습니다.

사용자가 생성되면 사용자 인증 정보가 시크릿에 생성됩니다. 애플리케이션은 인증에 사용자 및 해당 자격 증명을 사용하고 메시지를 생성하거나 사용해야 합니다.

인증에 대한 인증 정보를 관리하는 것 외에도 User Operator는 KafkaUser 선언에 사용자의 액세스 권한에 대한 설명을 포함하여 권한 부여 규칙을 관리합니다.

AMQ Streams Operator는 기능 게이트를 사용하여 특정 기능 및 기능을 활성화하거나 비활성화합니다. 기능 게이트를 활성화하면 연결된 Operator의 동작이 변경되어 AMQ Streams 배포에 해당 기능이 도입됩니다.

기능 게이트는 Operator 구성에 설정되며 alpha, beta 또는 Grad의 3단계 완성도가 있습니다. 중개 기능 게이트는 GA(General Availability)에 도달했으며 영구적으로 활성화된 기능입니다.

자세한 내용은 기능 게이트 를 참조하십시오.

CRD를 설치하여 새 사용자 정의 리소스 유형을 클러스터에 추가한 후 사양에 따라 리소스의 인스턴스를 생성할 수 있습니다.

AMQ Streams 구성 요소의 사용자 정의 리소스에는 spec 아래에 정의된 공통 구성 속성이 있습니다.

Kafka 주제 사용자 정의 리소스의 이 조각에서 apiVersion 및 kind 속성은 관련 CRD를 식별합니다. spec 속성은 주제의 파티션 및 복제본 수를 정의하는 구성을 보여줍니다.

Kafka 주제 사용자 정의 리소스

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: my-topic
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 1
  replicas: 1
  # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: my-topic
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 1
  replicas: 1
  # ...

YAML 정의에 통합할 수 있는 몇 가지 일반적인 구성 옵션은 특정 구성 요소와 관련된 많은 추가 구성 옵션이 있습니다.

리소스에 공통된 일부 구성 옵션은 여기에 설명되어 있습니다. 보안 및 메트릭 수집 도 적용 가능한 경우 채택될 수 있습니다.

일반적인 구성을 표시하는 YAML의 예

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-cluster
spec:
  # ...
  bootstrapServers: my-cluster-kafka-bootstrap:9092
  resources:
    requests:
      cpu: 12
      memory: 64Gi
    limits:
      cpu: 12
      memory: 64Gi
  logging:
    type: inline
    loggers:
      connect.root.logger.level: INFO
  readinessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  jvmOptions:
    "-Xmx": "2g"
    "-Xms": "2g"
  template:
    pod:
      affinity:
        nodeAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            nodeSelectorTerms:
              - matchExpressions:
                  - key: node-type
                    operator: In
                    values:
                      - fast-network
  # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-cluster
spec:
  # ...
  bootstrapServers: my-cluster-kafka-bootstrap:9092
  resources:
    requests:
      cpu: 12
      memory: 64Gi
    limits:
      cpu: 12
      memory: 64Gi
  logging:
    type: inline
    loggers:
      connect.root.logger.level: INFO
  readinessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  jvmOptions:
    "-Xmx": "2g"
    "-Xms": "2g"
  template:
    pod:
      affinity:
        nodeAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            nodeSelectorTerms:
              - matchExpressions:
                  - key: node-type
                    operator: In
                    values:
                      - fast-network
  # ...

kafka 클러스터는 하나 이상의 브로커로 구성됩니다. 생산자와 소비자가 브로커 내의 주제에 액세스할 수 있으려면 Kafka 구성에서 클러스터에 데이터가 저장되는 방법과 데이터에 액세스하는 방법을 정의해야 합니다. 랙에서 여러 브로커 노드로 실행되도록 Kafka 클러스터를 구성할 수 있습니다.

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
      - name: external1
        port: 9094
        type: route
        tls: true
        authentication:
          type: tls
    # ...
    storage:
      type: persistent-claim
      size: 10000Gi
    # ...
    rack:
      topologyKey: topology.kubernetes.io/zone
    config:
      replica.selector.class: org.apache.kafka.common.replica.RackAwareReplicaSelector
    # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
      - name: external1
        port: 9094
        type: route
        tls: true
        authentication:
          type: tls
    # ...
    storage:
      type: persistent-claim
      size: 10000Gi
    # ...
    rack:
      topologyKey: topology.kubernetes.io/zone
    config:
      replica.selector.class: org.apache.kafka.common.replica.RackAwareReplicaSelector
    # ...

노드 풀은 Kafka 클러스터 내의 별도의 Kafka 노드 그룹을 나타냅니다. 노드 풀을 사용하면 노드가 동일한 Kafka 클러스터 내에서 다른 구성을 가질 수 있습니다. 노드 풀에 지정되지 않은 구성 옵션은 Kafka 구성에서 상속됩니다.

하나 이상의 노드 풀을 사용하여 Kafka 클러스터를 배포할 수 있습니다. 노드 풀 구성에는 필수 및 선택적 설정이 포함됩니다. 복제본, 역할 및 스토리지에 대한 구성이 필요합니다.

KRaft 모드를 사용하는 경우 노드 풀의 모든 노드가 브로커, 컨트롤러 또는 둘 다로 작동하는 역할을 지정할 수 있습니다. 컨트롤러 및 이중 역할은 KRaft에 따라 다릅니다. 클러스터 관리에 Zoo Cryostat를 사용하는 Kafka 클러스터를 사용하는 경우 브로커 역할로 구성된 노드 풀을 사용할 수 있습니다.

노드 풀 구성을 표시하는 YAML의 예

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: pool-a
  labels:
    strimzi.io/cluster: my-cluster
spec:
  replicas: 3
  roles:
    - broker
  storage:
    type: jbod
    volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        deleteClaim: false

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: pool-a
  labels:
    strimzi.io/cluster: my-cluster
spec:
  replicas: 3
  roles:
    - broker
  storage:
    type: jbod
    volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        deleteClaim: false

Kafka MirrorMaker 2는 데이터 센터 내부 또는 여러 개의 활성 Kafka 클러스터 간에 데이터를 복제합니다. MirrorMaker 2를 설정하려면 소스 및 대상(대상) Kafka 클러스터가 실행 중이어야 합니다.

소스 클러스터에서 대상 클러스터로 데이터를 미러링 하는 프로세스는 비동기적입니다. 각 MirrorMaker 2 인스턴스는 하나의 소스 클러스터에서 하나의 대상 클러스터로 데이터를 미러링합니다. 두 개 이상의 MirrorMaker 2 인스턴스를 사용하여 클러스터 간에 데이터를 미러링할 수 있습니다.

그림 7.1. 두 클러스터에서 복제

MirrorMaker 2는 다음과 같이 소스 및 대상 클러스터 구성을 사용합니다.

주제 및 소비자 그룹 복제는 쉼표로 구분된 목록 또는 정규식 패턴으로 지정됩니다.

MirrorMaker 2 구성을 표시하는 YAML의 예

  apiVersion: kafka.strimzi.io/v1beta2
  kind: KafkaMirrorMaker2
  metadata:
    name: my-mirror-maker2
    spec:
      version: 3.7.0
      connectCluster: "my-cluster-target"
      clusters:
      - alias: "my-cluster-source"
        bootstrapServers: my-cluster-source-kafka-bootstrap:9092
      - alias: "my-cluster-target"
        bootstrapServers: my-cluster-target-kafka-bootstrap:9092
      mirrors:
      - sourceCluster: "my-cluster-source"
        targetCluster: "my-cluster-target"
        sourceConnector: {}
      topicsPattern: ".*"
      groupsPattern: "group1|group2|group3"

  apiVersion: kafka.strimzi.io/v1beta2
  kind: KafkaMirrorMaker2
  metadata:
    name: my-mirror-maker2
    spec:
      version: 3.7.0
      connectCluster: "my-cluster-target"
      clusters:
      - alias: "my-cluster-source"
        bootstrapServers: my-cluster-source-kafka-bootstrap:9092
      - alias: "my-cluster-target"
        bootstrapServers: my-cluster-target-kafka-bootstrap:9092
      mirrors:
      - sourceCluster: "my-cluster-source"
        targetCluster: "my-cluster-target"
        sourceConnector: {}
      topicsPattern: ".*"
      groupsPattern: "group1|group2|group3"

Kafka MirrorMaker(MirrorMaker라고도 함)는 생산자와 소비자를 사용하여 다음과 같이 클러스터 간에 데이터를 복제합니다.

소비자 및 생산자 구성에는 필요한 인증 및 암호화 설정이 포함됩니다. include 속성은 소스에서 대상 클러스터로 미러링할 주제를 정의합니다.

주요 소비자 구성

주요 Producer 구성

MirrorMaker 구성을 표시하는 YAML의 예

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker
metadata:
  name: my-mirror-maker
spec:
  # ...
  consumer:
    bootstrapServers: my-source-cluster-kafka-bootstrap:9092
    groupId: "my-group"
    numStreams: 2
    offsetCommitInterval: 120000
    # ...
  producer:
    # ...
    abortOnSendFailure: false
    # ...
  include: "my-topic|other-topic"
  # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker
metadata:
  name: my-mirror-maker
spec:
  # ...
  consumer:
    bootstrapServers: my-source-cluster-kafka-bootstrap:9092
    groupId: "my-group"
    numStreams: 2
    offsetCommitInterval: 120000
    # ...
  producer:
    # ...
    abortOnSendFailure: false
    # ...
  include: "my-topic|other-topic"
  # ...

AMQ Streams의 KafkaConnect 리소스를 사용하여 새 Kafka Connect 클러스터를 빠르고 쉽게 생성합니다.

KafkaConnect 리소스를 사용하여 Kafka Connect 를 배포할 때 Kafka 클러스터에 연결하기 위해 부트스트랩 서버 주소 ( spec.bootstrapServers)를 지정합니다. 서버가 다운된 경우 두 개 이상의 주소를 지정할 수 있습니다. 또한 인증 자격 증명 및 TLS 암호화 인증서를 지정하여 보안 연결을 만듭니다.

KafkaConnect 리소스를 사용하여 다음을 지정할 수도 있습니다.

Cluster Operator는 KafkaConnector 리소스를 사용하여 생성된 KafkaConnect 리소스 및 커넥터를 사용하여 배포된 Kafka Connect 클러스터를 관리합니다.

플러그인 구성

플러그인은 커넥터 인스턴스를 생성하기 위한 구현을 제공합니다. 플러그인이 인스턴스화되면 특정 유형의 외부 데이터 시스템에 연결하기 위한 구성이 제공됩니다. 플러그인은 지정된 종류의 데이터 소스에 연결하기 위한 커넥터 및 작업 구현을 정의하는 하나 이상의 JAR 파일 세트를 제공합니다. 많은 외부 시스템에 대한 플러그인은 Kafka Connect에서 사용할 수 있습니다. 고유한 플러그인을 만들 수도 있습니다.

이 구성은 Kafka Connect 내외로 피드할 소스 입력 데이터 및 대상 출력 데이터를 설명합니다. 소스 커넥터의 경우 외부 소스 데이터는 메시지를 저장할 특정 주제를 참조해야 합니다. 이 플러그인은 데이터를 변환하는 데 필요한 라이브러리 및 파일을 포함할 수도 있습니다.

Kafka Connect 배포에는 플러그인이 하나 이상 있을 수 있지만 각 플러그인마다 하나의 버전만 있을 수 있습니다.

선택한 플러그인이 포함된 사용자 지정 Kafka Connect 이미지를 생성할 수 있습니다. 다음 두 가지 방법으로 이미지를 생성할 수 있습니다.

컨테이너 이미지를 자동으로 생성하려면 KafkaConnect 리소스의 build 속성을 사용하여 Kafka Connect 클러스터에 추가할 플러그인을 지정합니다. AMQ Streams는 플러그인 아티팩트를 자동으로 다운로드하고 새 컨테이너 이미지에 추가합니다.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  # ...
  build: 
    output: 
      type: docker
      image: my-registry.io/my-org/my-connect-cluster:latest
      pushSecret: my-registry-credentials
    plugins: 
      - name: my-connector
        artifacts:
          - type: tgz
            url: https://<plugin_download_location>.tgz
            sha512sum: <checksum_to_verify_the_plugin>
      # ...
  # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  # ...
  build: 

    output: 

      type: docker
      image: my-registry.io/my-org/my-connect-cluster:latest
      pushSecret: my-registry-credentials
    plugins: 

      - name: my-connector
        artifacts:
          - type: tgz
            url: https://<plugin_download_location>.tgz
            sha512sum: <checksum_to_verify_the_plugin>
      # ...
  # ...

Dockerfile을 사용하여 이미지를 빌드하는 경우 AMQ Streams의 최신 컨테이너 이미지를 기본 이미지로 사용하여 플러그인 구성 파일을 추가할 수 있습니다.

FROM registry.redhat.io/amq-streams/kafka-37-rhel8:2.7.0
USER root:root
COPY ./my-plugins/ /opt/kafka/plugins/
USER 1001

FROM registry.redhat.io/amq-streams/kafka-37-rhel8:2.7.0
USER root:root
COPY ./my-plugins/ /opt/kafka/plugins/
USER 1001

작업자에 대한 Kafka Connect 클러스터 구성

KafkaConnect 리소스의 config 속성에 작업자의 구성을 지정합니다.

분산 Kafka Connect 클러스터에는 그룹 ID와 내부 구성 주제 세트가 있습니다.

Kafka Connect 클러스터는 기본적으로 이러한 속성에 대해 동일한 값으로 구성됩니다. Kafka Connect 클러스터는 오류를 생성하기 때문에 그룹 ID 또는 주제 이름을 공유할 수 없습니다. 서로 다른 Kafka Connect 클러스터를 여러 개 사용하는 경우 생성된 각 Kafka Connect 클러스터의 작업자에 대해 이러한 설정을 고유해야 합니다.

각 Kafka Connect 클러스터에서 사용하는 커넥터의 이름도 고유해야 합니다.

다음 예제 작업자 구성에서는 JSON 변환기가 지정됩니다. Kafka Connect에서 사용하는 내부 Kafka 항목에 대한 복제 요인이 설정됩니다. 프로덕션 환경에는 최소 3개 이상 있어야 합니다. 주제를 만든 후 복제 요소를 변경하면 적용되지 않습니다.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
# ...
spec:
  config:
    # ...
    group.id: my-connect-cluster 
    offset.storage.topic: my-connect-cluster-offsets 
    config.storage.topic: my-connect-cluster-configs 
    status.storage.topic: my-connect-cluster-status 
    key.converter: org.apache.kafka.connect.json.JsonConverter 
    value.converter: org.apache.kafka.connect.json.JsonConverter 
    key.converter.schemas.enable: true 
    value.converter.schemas.enable: true 
    config.storage.replication.factor: 3 
    offset.storage.replication.factor: 3 
    status.storage.replication.factor: 3 
  # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
# ...
spec:
  config:
    # ...
    group.id: my-connect-cluster 

    offset.storage.topic: my-connect-cluster-offsets 

    config.storage.topic: my-connect-cluster-configs 

    status.storage.topic: my-connect-cluster-status 

    key.converter: org.apache.kafka.connect.json.JsonConverter 

    value.converter: org.apache.kafka.connect.json.JsonConverter 

    key.converter.schemas.enable: true 

    value.converter.schemas.enable: true 

    config.storage.replication.factor: 3 

    offset.storage.replication.factor: 3 

    status.storage.replication.factor: 3 

  # ...

커넥터의 KafkaConnector 관리

배포의 작업자 Pod에 사용되는 컨테이너 이미지에 플러그인을 추가한 후 AMQ Streams의 KafkaConnector 사용자 정의 리소스 또는 Kafka Connect API를 사용하여 커넥터 인스턴스를 관리할 수 있습니다. 이러한 옵션을 사용하여 새 커넥터 인스턴스를 생성할 수도 있습니다.

KafkaConnector 리소스는 Cluster Operator의 커넥터 관리에 대한 OpenShift 네이티브 접근 방식을 제공합니다. KafkaConnector 리소스로 커넥터를 관리하려면 KafkaConnect 사용자 정의 리소스에 주석을 지정해야 합니다.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
  # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
  # ...

use-connector-resources 를 true 로 설정하면 KafkaConnectors가 커넥터를 생성, 삭제 및 재구성할 수 있습니다.

KafkaConnect 구성에서 use-connector-resources 가 활성화된 경우 KafkaConnector 리소스를 사용하여 커넥터를 정의하고 관리해야 합니다. KafkaConnector 리소스는 외부 시스템에 연결하도록 구성됩니다. 외부 데이터 시스템과 상호 작용하는 Kafka Connect 클러스터 및 Kafka 클러스터와 동일한 OpenShift 클러스터에 배포됩니다.

구성은 커넥터 인스턴스가 인증을 포함하여 외부 데이터 시스템에 연결하는 방법을 지정합니다. 또한 볼 데이터를 설명해야 합니다. 소스 커넥터의 경우 구성에 데이터베이스 이름을 지정할 수 있습니다. 대상 주제 이름을 지정하여 Kafka에 데이터가 있어야 하는 위치를 지정할 수도 있습니다.

tasksMax 를 사용하여 최대 작업 수를 지정합니다. 예를 들어 tasksMax: 2 가 있는 소스 커넥터는 소스 데이터 가져오기를 두 작업으로 분할할 수 있습니다.

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 
  autoRestart: 
    enabled: true
  config: 
    file: "/opt/kafka/LICENSE" 
    topic: my-topic 
    # ...

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 

  autoRestart: 

    enabled: true
  config: 

    file: "/opt/kafka/LICENSE" 

    topic: my-topic 

    # ...

Kafka 연결 API

Kafka Connect REST API를 KafkaConnector 리소스를 사용하여 커넥터를 관리하는 대신 사용합니다. Kafka Connect REST API는 < connect_cluster_name> -connect-api:8083 에서 실행되는 서비스로 사용할 수 있습니다. 여기서 < connect_cluster_name >은 Kafka Connect 클러스터의 이름입니다.

커넥터 구성을 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"
    }
}'

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"
    }
}'

KafkaConnectors가 활성화된 경우 Cluster Operator에 의해 Kafka Connect REST API를 사용하여 직접 변경한 수동 변경 사항을 되돌립니다.

REST API에서 지원하는 작업은 Apache Kafka Connect API 설명서에 설명되어 있습니다.

Kafka Bridge 구성에는 연결하는 Kafka 클러스터의 부트스트랩 서버 사양과 필요한 모든 암호화 및 인증 옵션이 필요합니다.

Kafka 브리지 소비자 및 생산자 구성은 생산자의 소비자 및 Apache Kafka 구성 설명서에 대한 Apache Kafka 구성 문서에 설명된 대로 표준입니다.

HTTP 관련 구성 옵션은 서버가 수신 대기하는 포트 연결을 설정합니다.

CORS

Kafka 브리지는 CORS(Cross-Origin Resource Sharing) 사용을 지원합니다. CORS는 브라우저에서 두 개 이상의 원본에서 선택한 리소스에 액세스할 수 있는 HTTP 메커니즘입니다(예: 다른 도메인의 리소스). CORS를 사용하도록 선택하는 경우 Kafka 브리지를 통해 Kafka 클러스터와 상호 작용하기 위해 허용된 리소스 원본 및 HTTP 메서드 목록을 정의할 수 있습니다. 목록은 Kafka 브리지 구성의 http 사양에 정의되어 있습니다.

CORS를 사용하면 다른 도메인의 원본 소스 간에 간단하고 미리 제공되는 요청을 수행할 수 있습니다.

Kafka 브리지 구성을 표시하는 YAML의 예

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaBridge
metadata:
  name: my-bridge
spec:
  # ...
  bootstrapServers: my-cluster-kafka:9092
  http:
    port: 8080
    cors:
      allowedOrigins: "https://strimzi.io"
      allowedMethods: "GET,POST,PUT,DELETE,OPTIONS,PATCH"
  consumer:
    config:
      auto.offset.reset: earliest
  producer:
    config:
      delivery.timeout.ms: 300000
  # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaBridge
metadata:
  name: my-bridge
spec:
  # ...
  bootstrapServers: my-cluster-kafka:9092
  http:
    port: 8080
    cors:
      allowedOrigins: "https://strimzi.io"
      allowedMethods: "GET,POST,PUT,DELETE,OPTIONS,PATCH"
  consumer:
    config:
      auto.offset.reset: earliest
  producer:
    config:
      delivery.timeout.ms: 300000
  # ...

AMQ Streams는 암호화된 통신을 위한 프로토콜인 TLS(Transport Layer Security)를 지원합니다.

통신은 항상 간의 통신을 위해 암호화됩니다.

Kafka 브로커와 클라이언트 간에 TLS 암호화를 구성할 수도 있습니다. Kafka 브로커의 외부 리스너를 구성할 때 외부 클라이언트에 대해 TLS가 지정됩니다.

AMQ Streams 구성 요소 및 Kafka 클라이언트는 암호화를 위해 디지털 인증서를 사용합니다. Cluster Operator는 Kafka 클러스터 내에서 암호화를 사용하도록 인증서를 설정합니다. Kafka 리스너 인증서라고 하는 자체 서버 인증서를 Kafka 클라이언트와 Kafka 브로커 간 통신 및 클러스터 간 통신에 제공할 수 있습니다.

AMQ Streams는 Secrets 를 사용하여 mTLS에 필요한 인증서 및 개인 키를 PEM 및 PKCS #12 형식으로 저장합니다.

TLS CA(인증 기관)는 구성 요소의 ID를 인증하기 위해 인증서를 발행합니다. AMQ Streams는 CA 인증서에 대해 구성 요소의 인증서를 확인합니다.

Kafka 리스너는 인증을 사용하여 Kafka 클러스터에 대한 보안 클라이언트 연결을 확인합니다.

지원되는 인증 메커니즘:

사용자 Operator는 OAuth 2.0이 아닌 mTLS 및 SCRAM 인증에 대한 사용자 인증 정보를 관리합니다. 예를 들어 User Operator를 통해 Kafka 클러스터에 액세스해야 하는 클라이언트를 나타내는 사용자를 생성하고 tls 를 인증 유형으로 지정할 수 있습니다.

OAuth 2.0 토큰 기반 인증을 사용하여 애플리케이션 클라이언트는 계정 인증 정보를 노출하지 않고 Kafka 브로커에 액세스할 수 있습니다. 권한 부여 서버는 액세스 권한 부여 및 액세스 관련 문의를 처리합니다.

사용자 정의 인증을 사용하면 Kafka에서 지원하는 모든 유형의 인증을 사용할 수 있습니다. 더 많은 유연성을 제공할 수 있지만 복잡성도 추가할 수 있습니다.

Kafka 클러스터는 권한을 사용하여 특정 클라이언트 또는 사용자가 Kafka 브로커에서 허용되는 작업을 제어합니다. Kafka 클러스터에 적용되는 경우 클라이언트 연결에 사용되는 모든 리스너에 대해 권한 부여가 활성화됩니다.

사용자가 Kafka 브로커 구성의 슈퍼 사용자 목록에 추가되면 권한 부여 메커니즘을 통해 구현된 권한 부여 제약 조건에 관계없이 클러스터에 대한 무제한 액세스가 허용됩니다.

지원되는 권한 부여 메커니즘:

간단한 인증에서는 다양한 리소스에 대한 사용자 액세스를 지정하는 ACL(액세스 제어 목록)을 관리하는 AclAuthorizer 및 StandardAuthorizer Kafka 플러그인을 사용합니다. 사용자 정의 권한 부여의 경우 ACL 규칙을 적용하도록 자체 Authorizer 플러그인을 구성합니다.

OAuth 2.0 및 OPA는 권한 부여 서버에서 정책 기반 제어를 제공합니다. Kafka 브로커의 리소스에 대한 액세스 권한을 부여하는 데 사용되는 보안 정책 및 권한은 권한 부여 서버에 정의되어 있습니다.

URL은 권한 부여 서버에 연결하고 클라이언트 또는 사용자가 요청한 작업이 허용 또는 거부되었는지 확인하는 데 사용됩니다. 사용자와 클라이언트는 Kafka 브로커에 대한 특정 작업을 수행하기 위해 액세스를 허용하는 권한 부여 서버에서 생성된 정책과 일치합니다.

연방 정보 처리 표준 (FIPS)은 정보 시스템에서 처리하거나 전송하는 민감한 데이터 및 정보의 기밀성, 무결성 및 가용성을 보장하기 위해 미국 정부에 의해 설정된 일련의 보안 표준입니다. AMQ Streams 컨테이너 이미지에서 사용되는 OpenJDK는 FIPS 지원 OpenShift 클러스터에서 실행할 때 FIPS 모드를 자동으로 활성화합니다.

NIST 검증 프로그램 및 검증된 모듈에 대한 자세한 내용은 NIST 웹 사이트의 암호화 모듈 유효성 검사 프로그램을 참조하십시오.

Prometheus는 Kafka 구성 요소 및 AMQ Streams Operator에서 메트릭 데이터를 추출할 수 있습니다.

Prometheus를 사용하여 지표 데이터를 가져오고 경고를 제공하려면 Prometheus 및 Prometheus Alertmanager 플러그인을 배포해야 합니다. 지표 데이터를 노출하려면 Kafka 리소스도 지표 구성으로 배포하거나 재배포해야 합니다.

Prometheus는 모니터링을 위해 노출된 지표 데이터를 스크랩합니다. 조건이 사전 정의된 경고 규칙에 따라 잠재적인 문제를 나타내는 경우 Alertmanager를 발행합니다.

샘플 메트릭 및 경고 규칙 구성 파일은 AMQ Streams와 함께 제공됩니다. AMQ Streams와 함께 제공되는 샘플 경고 메커니즘은 Slack 채널에 알림을 전송하도록 구성되어 있습니다.

Grafana는 Prometheus에서 노출하는 메트릭 데이터를 사용하여 모니터링을 위한 대시보드 시각화를 제공합니다.

Prometheus가 데이터 소스로 추가된 Grafana 배포가 필요합니다. AMQ Streams에 JSON 파일로 제공된 대시보드의 예는 Grafana 인터페이스를 통해 모니터링 데이터를 제공합니다.

Kafka Exporter는 Apache Kafka 브로커 및 클라이언트의 모니터링을 개선하는 오픈 소스 프로젝트입니다. Kafka 내보내기는 Kafka 클러스터와 함께 배포하여 오프셋, 소비자 그룹, 소비자 지연 및 주제와 관련된 Kafka 브로커에서 추가 Prometheus 지표 데이터를 추출합니다. 제공된 Grafana 대시보드를 사용하여 Kafka Exporter에서 Prometheus에서 수집한 데이터를 시각화할 수 있습니다.

Kafka Exporter의 샘플 구성 파일, 경고 규칙 및 Grafana 대시보드는 AMQ Streams와 함께 제공됩니다.

분산 추적은 분산 시스템의 애플리케이션 간 트랜잭션 진행 상황을 추적합니다. 마이크로 서비스 아키텍처에서 추적은 서비스 간 트랜잭션 진행 상황을 추적합니다. 추적 데이터는 애플리케이션 성능을 모니터링하고 대상 시스템 및 최종 사용자 애플리케이션 관련 문제를 조사하는 데 유용합니다.

AMQ Streams에서 추적은 소스 시스템에서 Kafka로 보낸 다음 Kafka에서 대상 시스템 및 애플리케이션까지 메시지의 엔드 투 엔드 추적을 용이하게 합니다. 분산 추적은 Grafana 대시보드의 메트릭 모니터링과 구성 요소 로거를 보완합니다.

추적 지원은 다음 Kafka 구성 요소에 빌드됩니다.

Kafka 브로커에서는 추적이 지원되지 않습니다.

사용자 정의 리소스를 통해 이러한 구성 요소에 대한 추적을 활성화하고 구성합니다. spec.template 속성을 사용하여 추적 구성을 추가합니다.

spec.tracing.type 속성을 사용하여 추적 유형을 지정하여 추적을 활성화합니다.

Kafka 클라이언트 추적

Kafka 생산자 및 소비자와 같은 클라이언트 애플리케이션도 트랜잭션을 모니터링하도록 설정할 수 있습니다. 클라이언트는 추적 프로필을 사용하여 구성되며 클라이언트 애플리케이션이 사용할 추적기가 초기화됩니다.

cruise Control은 다음 Kafka 작업을 지원하는 오픈 소스 시스템입니다.

이 작업은 브로커 Pod를 보다 효율적으로 사용하는 균형 있는 Kafka 클러스터를 실행하는 데 도움이 됩니다.

일반적인 클러스터는 시간이 지남에 따라 균등하게 로드할 수 있습니다. 대량의 메시지 트래픽을 처리하는 파티션은 사용 가능한 브로커에 균등하게 분산되지 않을 수 있습니다. 클러스터를 재조정하려면 관리자가 브로커의 부하를 모니터링하고 사용 중인 파티션을 예비 용량이 있는 브로커에 수동으로 다시 할당해야 합니다.

cruise Control은 클러스터 재조정 프로세스를 자동화합니다. CPU, 디스크 및 네트워크 로드를 기반으로 클러스터에 대한 리소스 사용률의 워크로드 모델을 구성하고, 분산 파티션 할당을 위해 최적화 제안(승인 또는 거부할 수 있음)을 생성합니다. 구성 가능한 최적화 목표 세트가 이러한 제안을 계산하는 데 사용됩니다.

특정 모드에서 최적화 제안을 생성할 수 있습니다. 기본 전체 모드는 모든 브로커에서 파티션을 재조정합니다. add-brokers 및 remove-brokers 모드를 사용하여 클러스터를 확장하거나 축소할 때 변경 사항을 적용할 수도 있습니다.

최적화 제안을 승인하면 Cruise Control이 Kafka 클러스터에 적용합니다. KafkaRebalance 리소스를 사용하여 최적화 제안을 구성하고 생성합니다. 최적화 제안이 자동으로 또는 수동으로 승인되도록 주석을 사용하여 리소스를 구성할 수 있습니다.