OpenShift에서 Apache Kafka용 스트림 배포 및 관리

프로세스

1. Cluster Operator를 설치할 네임스페이스를 사용하도록 Apache Kafka 설치 파일의 Streams를 편집합니다.

   예를 들어 이 절차에서는 Cluster Operator가 네임스페이스 my-cluster-operator-namespace 에 설치됩니다.

   Linux에서 다음을 사용하십시오.

   sed -i 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

   MacOS에서 다음을 사용하십시오.

   sed -i '' 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

2. Cluster Operator를 배포합니다.

   oc create -f install/cluster-operator -n my-cluster-operator-namespace

3. 배포 상태를 확인합니다.

   oc get deployments -n my-cluster-operator-namespace

   출력에 배포 이름 및 준비 상태 표시

   NAME                      READY  UP-TO-DATE  AVAILABLE
   strimzi-cluster-operator  1/1    1           1

   READY 는 ready/expected 복제본 수를 표시합니다. AVAILABLE 출력에 1 이 표시되면 배포가 성공적으로 수행됩니다.

프로세스

1. Cluster Operator를 설치할 네임스페이스를 사용하도록 Apache Kafka 설치 파일의 Streams를 편집합니다.

   예를 들어 이 절차에서는 Cluster Operator가 네임스페이스 my-cluster-operator-namespace 에 설치됩니다.

   Linux에서 다음을 사용하십시오.

   sed -i 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

   MacOS에서 다음을 사용하십시오.

   sed -i '' 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

2. install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml 파일을 편집하여 Cluster Operator가 STRIMZI_NAMESPACE 환경 변수에 감시하는 모든 네임스페이스 목록을 추가합니다.

   예를 들어 이 절차에서 Cluster Operator는 네임스페이스 watched-namespace-1,watched-namespace-2,watched-namespace-3 을 확인합니다.

   apiVersion: apps/v1
   kind: Deployment
   spec:
     # ...
     template:
       spec:
         serviceAccountName: strimzi-cluster-operator
         containers:
         - name: strimzi-cluster-operator
           image: registry.redhat.io/amq-streams/strimzi-rhel9-operator:2.7.0
           imagePullPolicy: IfNotPresent
           env:
           - name: STRIMZI_NAMESPACE
             value: watched-namespace-1,watched-namespace-2,watched-namespace-3

3. 나열된 각 네임스페이스에 대해 RoleBinding을 설치합니다.

   이 예제에서는 이러한 명령에서 watched-namespace -namespace를 이전 단계에 나열된 네임스페이스로 교체하여 watched-namespace-1, watched-namespace -2,watched-namespace-3:을 반복합니다.

   oc create -f install/cluster-operator/020-RoleBinding-strimzi-cluster-operator.yaml -n <watched_namespace>
   oc create -f install/cluster-operator/023-RoleBinding-strimzi-cluster-operator.yaml -n <watched_namespace>
   oc create -f install/cluster-operator/031-RoleBinding-strimzi-cluster-operator-entity-operator-delegation.yaml -n <watched_namespace>

4. Cluster Operator를 배포합니다.

   oc create -f install/cluster-operator -n my-cluster-operator-namespace

5. 배포 상태를 확인합니다.

   oc get deployments -n my-cluster-operator-namespace

   출력에 배포 이름 및 준비 상태 표시

   NAME                      READY  UP-TO-DATE  AVAILABLE
   strimzi-cluster-operator  1/1    1           1

   READY 는 ready/expected 복제본 수를 표시합니다. AVAILABLE 출력에 1 이 표시되면 배포가 성공적으로 수행됩니다.

프로세스

1. Cluster Operator를 설치할 네임스페이스를 사용하도록 Apache Kafka 설치 파일의 Streams를 편집합니다.

   예를 들어 이 절차에서는 Cluster Operator가 네임스페이스 my-cluster-operator-namespace 에 설치됩니다.

   Linux에서 다음을 사용하십시오.

   sed -i 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

   MacOS에서 다음을 사용하십시오.

   sed -i '' 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

2. install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml 파일을 편집하여 STRIMZI_NAMESPACE 환경 변수의 값을 * 로 설정합니다.

   apiVersion: apps/v1
   kind: Deployment
   spec:
     # ...
     template:
       spec:
         # ...
         serviceAccountName: strimzi-cluster-operator
         containers:
         - name: strimzi-cluster-operator
           image: registry.redhat.io/amq-streams/strimzi-rhel9-operator:2.7.0
           imagePullPolicy: IfNotPresent
           env:
           - name: STRIMZI_NAMESPACE
             value: "*"
           # ...

3. Cluster Operator에 모든 네임스페이스에 대한 클러스터 전체 액세스 권한을 부여하는 ClusterRoleBindings 를 생성합니다.

   oc create clusterrolebinding strimzi-cluster-operator-namespaced --clusterrole=strimzi-cluster-operator-namespaced --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator
   oc create clusterrolebinding strimzi-cluster-operator-watched --clusterrole=strimzi-cluster-operator-watched --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator
   oc create clusterrolebinding strimzi-cluster-operator-entity-operator-delegation --clusterrole=strimzi-entity-operator --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator

4. OpenShift 클러스터에 Cluster Operator를 배포합니다.

   oc create -f install/cluster-operator -n my-cluster-operator-namespace

5. 배포 상태를 확인합니다.

   oc get deployments -n my-cluster-operator-namespace

   출력에 배포 이름 및 준비 상태 표시

   NAME                      READY  UP-TO-DATE  AVAILABLE
   strimzi-cluster-operator  1/1    1           1

   READY 는 ready/expected 복제본 수를 표시합니다. AVAILABLE 출력에 1 이 표시되면 배포가 성공적으로 수행됩니다.

프로세스

1. KRaft 기반 Kafka 클러스터를 배포합니다.

   • 이중 역할 노드를 사용하는 단일 노드 풀을 사용하여 KRaft 모드에서 Kafka 클러스터를 배포하려면 다음을 수행합니다.

     oc apply -f examples/kafka/kraft/kafka-with-dual-role-nodes.yaml

   • 브로커 및 컨트롤러 노드에 대해 별도의 노드 풀을 사용하여 KRaft 모드에서 영구 Kafka 클러스터를 배포하려면 다음을 수행합니다.

     oc apply -f examples/kafka/kraft/kafka.yaml

   • 브로커 및 컨트롤러 노드에 대해 별도의 노드 풀을 사용하여 KRaft 모드에서 임시 Kafka 클러스터를 배포하려면 다음을 수행합니다.

     oc apply -f examples/kafka/kraft/kafka-ephemeral.yaml

   • 3개의 브로커로 구성된 두 개의 노드 풀을 사용하여 Kafka 클러스터 및 Zoo Cryostat 클러스터를 배포하려면 다음을 수행합니다.

     oc apply -f examples/kafka/kafka-with-node-pools.yaml

2. 배포 상태를 확인합니다.

   oc get pods -n <my_cluster_operator_namespace>

   출력에 노드 풀 이름 및 준비 상태가 표시됩니다.

   NAME                        READY  STATUS   RESTARTS
   my-cluster-entity-operator  3/3    Running  0
   my-cluster-pool-a-0         1/1    Running  0
   my-cluster-pool-a-1         1/1    Running  0
   my-cluster-pool-a-4         1/1    Running  0

   • my-cluster 는 Kafka 클러스터의 이름입니다.

   • pool-a 는 노드 풀의 이름입니다.

     0 으로 시작하는 순차적 인덱스 번호는 생성된 각 Kafka Pod를 식별합니다. Zoo Cryostat를 사용하는 경우 Zoo Cryostat Pod도 표시됩니다.

     READY 는 ready/expected 복제본 수를 표시합니다. STATUS 가 Running 으로 표시되면 배포가 성공적으로 수행됩니다.

     배포에 대한 정보는 풀에 있는 노드의 ID 목록을 포함하여 KafkaNodePool 리소스의 상태에도 표시됩니다.

     참고

     노드 ID는 클러스터 내의 모든 노드 풀에서 0(0)부터 순차적으로 할당됩니다. 즉, 노드 ID가 특정 노드 풀 내에서 순차적으로 실행되지 않을 수 있습니다. 클러스터 전체의 노드 ID 시퀀스에 차이가 있는 경우 추가할 다음 노드에는 격차를 채우는 ID가 할당됩니다. 축소하면 풀 내에서 노드 ID가 가장 높은 노드가 제거됩니다.

프로세스

1. Zoo Cryostat 기반 Kafka 클러스터를 배포합니다.

   • 임시 클러스터를 배포하려면 다음을 수행합니다.

     oc apply -f examples/kafka/kafka-ephemeral.yaml

   • 영구 클러스터를 배포하려면 다음을 수행합니다.

     oc apply -f examples/kafka/kafka-persistent.yaml

2. 배포 상태를 확인합니다.

   oc get pods -n <my_cluster_operator_namespace>

   출력에 Pod 이름 및 준비 상태 표시

   NAME                        READY   STATUS    RESTARTS
   my-cluster-entity-operator  3/3     Running   0
   my-cluster-kafka-0          1/1     Running   0
   my-cluster-kafka-1          1/1     Running   0
   my-cluster-kafka-2          1/1     Running   0
   my-cluster-zookeeper-0      1/1     Running   0
   my-cluster-zookeeper-1      1/1     Running   0
   my-cluster-zookeeper-2      1/1     Running   0

   my-cluster 는 Kafka 클러스터의 이름입니다.

   0 으로 시작하는 순차적 인덱스 번호는 생성된 각 Kafka 및 Zoo Cryostat Pod를 식별합니다.

   기본 배포를 통해 Entity Operator 클러스터, 3 Kafka Pod 및 3 Zoo Cryostat Pod를 생성합니다.

   READY 는 ready/expected 복제본 수를 표시합니다. STATUS 가 Running 으로 표시되면 배포가 성공적으로 수행됩니다.

프로세스

1. topicOperator 를 포함하도록 Kafka 리소스의 entityOperator 속성을 편집합니다.

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     #...
     entityOperator:
       topicOperator: {}
       userOperator: {}

2. EntityTopicOperatorSpec 스키마 참조에 설명된 속성을 사용하여 Topic Operator 사양 을 구성합니다.

   모든 속성에서 기본값을 사용하려면 빈 오브젝트({})를 사용합니다.

3. 리소스를 생성하거나 업데이트합니다.

   oc apply -f <kafka_configuration_file>

4. 배포 상태를 확인합니다.

   oc get pods -n <my_cluster_operator_namespace>

   출력에 Pod 이름 및 준비 상태 표시

   NAME                        READY   STATUS    RESTARTS
   my-cluster-entity-operator  3/3     Running   0
   # ...

   my-cluster 는 Kafka 클러스터의 이름입니다.

   READY 는 ready/expected 복제본 수를 표시합니다. STATUS 가 Running 으로 표시되면 배포가 성공적으로 수행됩니다.

프로세스

1. userOperator 를 포함하도록 Kafka 리소스의 entityOperator 속성을 편집합니다.

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     #...
     entityOperator:
       topicOperator: {}
       userOperator: {}

2. EntityUserOperatorSpec 스키마 참조에 설명된 속성을 사용하여 User Operator 사양 을 구성합니다.

   모든 속성에서 기본값을 사용하려면 빈 오브젝트({})를 사용합니다.

3. 리소스를 생성하거나 업데이트합니다.

   oc apply -f <kafka_configuration_file>

4. 배포 상태를 확인합니다.

   oc get pods -n <my_cluster_operator_namespace>

   출력에 Pod 이름 및 준비 상태 표시

   NAME                        READY   STATUS    RESTARTS
   my-cluster-entity-operator  3/3     Running   0
   # ...

   my-cluster 는 Kafka 클러스터의 이름입니다.

   READY 는 ready/expected 복제본 수를 표시합니다. STATUS 가 Running 으로 표시되면 배포가 성공적으로 수행됩니다.

프로세스

1. OpenShift 콘솔을 사용하여 터미널을 열거나 CLI에서 exec 명령을 실행합니다.

   예를 들면 다음과 같습니다.

   oc exec -ti my-cluster-zookeeper-0 -- bin/zookeeper-shell.sh localhost:12181 ls /

   localhost:12181 을 사용해야 합니다.

프로세스

1. OpenShift 클러스터에 Kafka Connect를 배포합니다. examples/connect/kafka-connect.yaml 파일을 사용하여 Kafka Connect를 배포합니다.

   oc apply -f examples/connect/kafka-connect.yaml

2. 배포 상태를 확인합니다.

   oc get pods -n <my_cluster_operator_namespace>

   출력에 배포 이름 및 준비 상태 표시

   NAME                                 READY  STATUS   RESTARTS
   my-connect-cluster-connect-<pod_id>  1/1    Running  0

   my-connect-cluster 는 Kafka Connect 클러스터의 이름입니다.

   Pod ID는 생성된 각 pod를 식별합니다.

   기본 배포를 사용하면 단일 Kafka Connect Pod를 생성합니다.

   READY 는 ready/expected 복제본 수를 표시합니다. STATUS 가 Running 으로 표시되면 배포가 성공적으로 수행됩니다.

프로세스

1. .spec.build.output 에 컨테이너 레지스트리를 지정하고 .spec.build.plugins: 추가 커넥터를 지정하여 KafkaConnect 사용자 지정 리소스를 구성합니다.

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
   spec: 

   1

   
     #...
     build:
       output: 

   2

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

   3

   
         - name: connector-1
           artifacts:
             - type: tgz
               url: <url_to_download_connector_1_artifact>
               sha512sum: <SHA-512_checksum_of_connector_1_artifact>
         - name: connector-2
           artifacts:
             - type: jar
               url: <url_to_download_connector_2_artifact>
               sha512sum: <SHA-512_checksum_of_connector_2_artifact>
     #...

   1

   Kafka Connect 클러스터의 사양 입니다.

   2

   (필수) 새 이미지를 내보내는 컨테이너 레지스트리의 구성입니다.

   3

   (필수) 새 컨테이너 이미지에 추가할 커넥터 플러그인 및 해당 아티팩트 목록입니다. 각 플러그인은 하나 이상의 아티팩트 로 구성해야 합니다.

2. 리소스를 생성하거나 업데이트합니다.

   $ oc apply -f <kafka_connect_configuration_file>

3. 새 컨테이너 이미지가 빌드되고 Kafka Connect 클러스터가 배포될 때까지 기다립니다.
4. 추가한 커넥터 플러그인을 사용하려면 Kafka Connect REST API 또는 KafkaConnector 사용자 정의 리소스를 사용합니다.

프로세스

1. registry.redhat.io/amq-streams/kafka-37-rhel9:2.7.0 을 기본 이미지로 사용하여 새 Dockerfile 을 생성합니다.

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

   플러그인 파일 예

   $ tree ./my-plugins/
   ./my-plugins/
   ├── debezium-connector-mongodb
   │   ├── bson-<version>.jar
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mongodb-<version>.jar
   │   ├── debezium-core-<version>.jar
   │   ├── LICENSE.txt
   │   ├── mongodb-driver-core-<version>.jar
   │   ├── README.md
   │   └── # ...
   ├── debezium-connector-mysql
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mysql-<version>.jar
   │   ├── debezium-core-<version>.jar
   │   ├── LICENSE.txt
   │   ├── mysql-binlog-connector-java-<version>.jar
   │   ├── mysql-connector-java-<version>.jar
   │   ├── README.md
   │   └── # ...
   └── debezium-connector-postgres
       ├── CHANGELOG.md
       ├── CONTRIBUTE.md
       ├── COPYRIGHT.txt
       ├── debezium-connector-postgres-<version>.jar
       ├── debezium-core-<version>.jar
       ├── LICENSE.txt
       ├── postgresql-<version>.jar
       ├── protobuf-java-<version>.jar
       ├── README.md
       └── # ...

   COPY 명령은 컨테이너 이미지에 복사할 플러그인 파일을 가리킵니다.

   이 예제에서는 Debezium 커넥터(MongoDB, MySQL 및 PostgreSQL)용 플러그인을 추가하지만 모든 파일이 간결하게 나열되지는 않습니다. Kafka Connect에서 실행되는 Debezium은 다른 Kafka Connect 작업과 동일합니다.

2. 컨테이너 이미지를 빌드합니다.
3. 사용자 정의 이미지를 컨테이너 레지스트리로 내보냅니다.

4. 새 컨테이너 이미지를 가리킵니다.

   다음 방법 중 하나로 이미지를 가리킬 수 있습니다.

   • KafkaConnect 사용자 정의 리소스의 KafkaConnect.spec.image 속성을 편집합니다.

     설정된 경우 이 속성은 Cluster Operator의 STRIMZI_KAFKA_CONNECT_IMAGES 환경 변수를 재정의합니다.

     apiVersion: kafka.strimzi.io/v1beta2
     kind: KafkaConnect
     metadata:
       name: my-connect-cluster
     spec: 

     1

     
       #...
       image: my-new-container-image 

     2

     
       config: 

     3

     
         #...

     1

     Kafka Connect 클러스터의 사양 입니다.

     2

     Kafka Connect Pod의 Docker 이미지입니다.

     3

     커넥터가 아닌 Kafka Connect 작업자 의 구성입니다.

   • install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml 파일에서 STRIMZI_KAFKA_CONNECT_IMAGES 환경 변수를 편집한 다음 Cluster Operator를 다시 설치합니다.

프로세스

1. 다음 방법 중 하나로 Kafka Connect에 FileStreamSourceConnector 및 FileStreamSinkConnector 플러그인을 추가합니다.

   • 플러그인으로 새 컨테이너 이미지를 자동으로 빌드하도록 Kafka 연결 구성
   • 기본 Kafka Connect 이미지에서 Docker 이미지 생성 (수동 또는 연속 통합을 사용)

2. Kafka Connect 구성에서 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"
   spec:
       # ...

   KafkaConnector 리소스가 활성화된 상태에서 Cluster Operator는 이를 감시합니다.

3. examples/connect/source-connector.yaml 파일을 편집합니다.

   KafkaConnector 소스 커넥터 구성 예

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-source-connector  

   1

   
     labels:
       strimzi.io/cluster: my-connect-cluster 

   2

   
   spec:
     class: org.apache.kafka.connect.file.FileStreamSourceConnector 

   3

   
     tasksMax: 2 

   4

   
     autoRestart: 

   5

   
       enabled: true
     config: 

   6

   
       file: "/opt/kafka/LICENSE" 

   7

   
       topic: my-topic 

   8

   
       # ...

   1

   커넥터의 이름으로 사용되는 KafkaConnector 리소스의 이름입니다. OpenShift 리소스에 유효한 모든 이름을 사용합니다.

   2

   커넥터 인스턴스를 생성하는 Kafka Connect 클러스터의 이름입니다. 커넥터는 연결된 Kafka Connect 클러스터와 동일한 네임스페이스에 배포해야 합니다.

   3

   커넥터 클래스의 전체 이름입니다. Kafka Connect 클러스터에서 사용하는 이미지에 있어야 합니다.

   4

   커넥터가 생성할 수 있는 최대 Kafka Connect 작업 수입니다.

   5

   실패한 커넥터 및 작업을 자동으로 다시 시작할 수 있습니다. 기본적으로 재시작 수는 indefinite이지만 maxRestarts 속성을 사용하여 자동 재시작 횟수에 대해 최대값을 설정할 수 있습니다.

   6

   연결을 키 -값 쌍으로 연결합니다.

   7

   외부 데이터 파일의 위치입니다. 이 예제에서는 /opt/kafka/LICENSE 파일에서 읽을 수 있도록 FileStreamSourceConnector 를 구성하고 있습니다.

   8

   소스 데이터를 게시하는 Kafka 주제입니다.

4. OpenShift 클러스터에서 소스 KafkaConnector 를 생성합니다.

   oc apply -f examples/connect/source-connector.yaml

5. examples/connect/sink-connector.yaml 파일을 생성합니다.

   touch examples/connect/sink-connector.yaml

6. 다음 YAML을 sink-connector.yaml 파일에 붙여넣습니다.

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: my-sink-connector
     labels:
       strimzi.io/cluster: my-connect
   spec:
     class: org.apache.kafka.connect.file.FileStreamSinkConnector 

   1

   
     tasksMax: 2
     config: 

   2

   
       file: "/tmp/my-file" 

   3

   
       topics: my-topic 

   4

   1

   커넥터 클래스의 전체 이름 또는 별칭입니다. Kafka Connect 클러스터에서 사용하는 이미지에 있어야 합니다.

   2

   연결을 키 -값 쌍으로 연결합니다.

   3

   소스 데이터를 게시하는 임시 파일입니다.

   4

   Kafka 주제에서 소스 데이터를 읽습니다.

7. OpenShift 클러스터에서 싱크 KafkaConnector 를 생성합니다.

   oc apply -f examples/connect/sink-connector.yaml

8. 커넥터 리소스가 생성되었는지 확인합니다.

   oc get kctr --selector strimzi.io/cluster=<my_connect_cluster> -o name
   
   my-source-connector
   my-sink-connector

   <my_connect_cluster>를 Kafka Connect 클러스터 이름으로 바꿉니다.

9. 컨테이너에서 kafka-console-consumer.sh 를 실행하여 소스 커넥터가 해당 항목에 기록된 메시지를 읽습니다.

   oc exec <my_kafka_cluster>-kafka-0 -i -t -- bin/kafka-console-consumer.sh --bootstrap-server <my_kafka_cluster>-kafka-bootstrap.NAMESPACE.svc:9092 --topic my-topic --from-beginning

   <my_kafka_cluster>를 Kafka 클러스터 이름으로 바꿉니다.

1. 구성과 함께 KafkaConnector 리소스를 배포하여 커넥터 인스턴스를 생성합니다.
2. strimzi.io/use-connector-resources 주석을 true 로 설정하여 Kafka Connect 구성에서 KafkaConnector 리소스를 활성화합니다.

프로세스

1. Kafka MirrorMaker를 OpenShift 클러스터에 배포합니다.

   MirrorMaker의 경우:

   oc apply -f examples/mirror-maker/kafka-mirror-maker.yaml

   MirrorMaker 2:

   oc apply -f examples/mirror-maker/kafka-mirror-maker-2.yaml

2. 배포 상태를 확인합니다.

   oc get pods -n <my_cluster_operator_namespace>

   출력에 배포 이름 및 준비 상태 표시

   NAME                                    READY  STATUS   RESTARTS
   my-mirror-maker-mirror-maker-<pod_id>   1/1    Running  1
   my-mm2-cluster-mirrormaker2-<pod_id>    1/1    Running  1

   my-mirror-maker 는 Kafka MirrorMaker 클러스터의 이름입니다. my-mm2-cluster 는 Kafka MirrorMaker 2 클러스터의 이름입니다.

   Pod ID는 생성된 각 pod를 식별합니다.

   기본 배포를 사용하면 단일 MirrorMaker 또는 MirrorMaker 2 Pod를 설치합니다.

   READY 는 ready/expected 복제본 수를 표시합니다. STATUS 가 Running 으로 표시되면 배포가 성공적으로 수행됩니다.

프로세스

1. OpenShift 클러스터에 Kafka 브리지를 배포합니다.

   oc apply -f examples/bridge/kafka-bridge.yaml

2. 배포 상태를 확인합니다.

   oc get pods -n <my_cluster_operator_namespace>

   출력에 배포 이름 및 준비 상태 표시

   NAME                       READY  STATUS   RESTARTS
   my-bridge-bridge-<pod_id>  1/1    Running  0

   my-bridge 는 Kafka Bridge 클러스터의 이름입니다.

   Pod ID는 생성된 각 pod를 식별합니다.

   기본 배포를 사용하면 단일 Kafka Bridge Pod를 설치합니다.

   READY 는 ready/expected 복제본 수를 표시합니다. STATUS 가 Running 으로 표시되면 배포가 성공적으로 수행됩니다.

프로세스

1. OpenShift 클러스터에 있는 포드 이름을 나열합니다.

   oc get pods -o name
   
   pod/kafka-consumer
   # ...
   pod/my-bridge-bridge-<pod_id>

2. 포트 8080 의 Kafka 브리지 Pod에 연결합니다.

   oc port-forward pod/my-bridge-bridge-<pod_id> 8080:8080 &

   참고

   로컬 시스템의 포트 8080이 이미 사용 중인 경우 8008 과 같은 대체 HTTP 포트를 사용합니다.

프로세스

1. install/topic-operator/05-Deployment-strimzi-operator.yaml 독립 실행형 배포 파일에서 env 속성을 편집합니다.

   독립 실행형 Topic Operator 배포 구성 예

   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: strimzi-topic-operator
     labels:
       app: strimzi
   spec:
     # ...
     template:
       # ...
       spec:
         # ...
         containers:
           - name: strimzi-topic-operator
             # ...
             env:
               - name: STRIMZI_NAMESPACE 

   1

   
                 valueFrom:
                   fieldRef:
                     fieldPath: metadata.namespace
               - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS 

   2

   
                 value: my-kafka-bootstrap-address:9092
               - name: STRIMZI_RESOURCE_LABELS 

   3

   
                 value: "strimzi.io/cluster=my-cluster"
               - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS 

   4

   
                 value: "120000"
               - name: STRIMZI_LOG_LEVEL 

   5

   
                 value: INFO
               - name: STRIMZI_TLS_ENABLED 

   6

   
                 value: "false"
               - name: STRIMZI_JAVA_OPTS 

   7

   
                 value: "-Xmx=512M -Xms=256M"
               - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 

   8

   
                 value: "-Djavax.net.debug=verbose -DpropertyName=value"
               - name: STRIMZI_PUBLIC_CA 

   9

   
                 value: "false"
               - name: STRIMZI_TLS_AUTH_ENABLED 

   10

   
                 value: "false"
               - name: STRIMZI_SASL_ENABLED 

   11

   
                 value: "false"
               - name: STRIMZI_SASL_USERNAME 

   12

   
                 value: "admin"
               - name: STRIMZI_SASL_PASSWORD 

   13

   
                 value: "password"
               - name: STRIMZI_SASL_MECHANISM 

   14

   
                 value: "scram-sha-512"
               - name: STRIMZI_SECURITY_PROTOCOL 

   15

   
                 value: "SSL"
               - name: STRIMZI_USE_FINALIZERS
                 value: "false" 

   16

   1

   Topic Operator에서 KafkaTopic 리소스를 조사하는 OpenShift 네임스페이스입니다. Kafka 클러스터의 네임스페이스를 지정합니다.

   2

   Kafka 클러스터의 모든 브로커를 검색하고 연결할 부트스트랩 브로커 주소의 호스트 및 포트 쌍입니다. 서버가 다운된 경우 쉼표로 구분된 목록을 사용하여 두 개 또는 세 개의 브로커 주소를 지정합니다.

   3

   Topic Operator에서 관리하는 KafkaTopic 리소스를 식별하는 레이블입니다. 이는 Kafka 클러스터의 이름일 필요는 없습니다. KafkaTopic 리소스에 할당된 레이블일 수 있습니다. Topic Operator를 두 개 이상 배포하는 경우 레이블은 각각 고유해야 합니다. 즉, 운영자는 동일한 리소스를 관리할 수 없습니다.

   4

   주기적 조정 간격(밀리초)입니다. 기본값은 120000 (2분)입니다.

   5

   로깅 메시지를 출력하는 수준입니다. 수준을 ERROR,WARNING,INFO,DEBUG 또는 TRACE 로 설정할 수 있습니다.

   6

   Kafka 브로커와 암호화된 통신에 대한 TLS 지원을 활성화합니다.

   7

   (선택 사항) Topic Operator를 실행하는 JVM에서 사용하는 Java 옵션입니다.

   8

   (선택 사항) Topic Operator에 설정된 디버깅(-D) 옵션.

   9

   (선택 사항) TLS가 STRIMZI_TLS_ENABLED 를 통해 활성화된 경우 신뢰 저장소 인증서 생성을 건너뜁니다. 이 환경 변수가 활성화된 경우 브로커는 TLS 인증서에 대해 신뢰할 수 있는 공개 인증 기관을 사용해야 합니다. 기본값은 false입니다.

   10

   (선택 사항) mTLS 인증을 위한 키 저장소 인증서를 생성합니다. 이를 false 로 설정하면 mTLS를 사용한 클라이언트 인증이 Kafka 브로커에 대해 비활성화됩니다. 기본값은 true입니다.

   11

   (선택 사항) Kafka 브로커에 연결할 때 클라이언트 인증에 대해 SASL 지원을 활성화합니다. 기본값은 false입니다.

   12

   (선택 사항) 클라이언트 인증에 대한 SASL 사용자 이름입니다. SASL이 STRIMZI_SASL_ENABLED 를 통해 활성화된 경우에만 필수입니다.

   13

   (선택 사항) 클라이언트 인증에 대한 SASL 암호입니다. SASL이 STRIMZI_SASL_ENABLED 를 통해 활성화된 경우에만 필수입니다.

   14

   (선택 사항) 클라이언트 인증을 위한 SASL 메커니즘입니다. SASL이 STRIMZI_SASL_ENABLED 를 통해 활성화된 경우에만 필수입니다. 값을 일반,scram-sha-256 또는 scram-sha-512 로 설정할 수 있습니다.

   15

   (선택 사항) Kafka 브로커와의 통신에 사용되는 보안 프로토콜입니다. 기본값은 "PLAINTEXT"입니다. 값을 PLAINTEXT,SSL,SASL_PLAINTEXT 또는 SASL_SSL 로 설정할 수 있습니다.

   16

   종료자를 사용하여 주제 삭제 를 제어하지 않으려면 STRIMZI_USE_FINALIZERS 를 false 로 설정합니다.

2. 공용 인증 기관의 인증서를 사용하는 Kafka 브로커에 연결하려면 STRIMZI_PUBLIC_CA 를 true 로 설정합니다. 예를 들어 Amazon AWS MSK 서비스를 사용하는 경우 이 속성을 true 로 설정합니다.

3. STRIMZI_TLS_ENABLED 환경 변수를 사용하여 mTLS를 활성화한 경우 Kafka 클러스터에 대한 연결을 인증하는 데 사용되는 키 저장소 및 신뢰 저장소를 지정합니다.

   mTLS 구성의 예

   # ....
   env:
     - name: STRIMZI_TRUSTSTORE_LOCATION 

   1

   
       value: "/path/to/truststore.p12"
     - name: STRIMZI_TRUSTSTORE_PASSWORD 

   2

   
       value: "TRUSTSTORE-PASSWORD"
     - name: STRIMZI_KEYSTORE_LOCATION 

   3

   
       value: "/path/to/keystore.p12"
     - name: STRIMZI_KEYSTORE_PASSWORD 

   4

   
       value: "KEYSTORE-PASSWORD"
   # ...

   1

   신뢰 저장소에는 Kafka 및 Zoo Cryostat 서버 인증서에 서명하는 데 사용되는 인증 기관의 공개 키가 포함되어 있습니다.

   2

   truststore에 액세스하기 위한 암호입니다.

   3

   키 저장소에는 mTLS 인증을 위한 개인 키가 포함되어 있습니다.

   4

   키 저장소에 액세스하기 위한 암호입니다.

4. 배포 구성에 변경 사항을 적용하여 Topic Operator를 배포합니다.

5. 배포 상태를 확인합니다.

   oc get deployments

   출력에 배포 이름 및 준비 상태 표시

   NAME                    READY  UP-TO-DATE  AVAILABLE
   strimzi-topic-operator  1/1    1           1

   READY 는 ready/expected 복제본 수를 표시합니다. AVAILABLE 출력에 1 이 표시되면 배포가 성공적으로 수행됩니다.

프로세스

1. install/user-operator/05-Deployment-strimzi-user-operator.yaml 독립 실행형 배포 파일에서 다음 env 속성을 편집합니다.

   독립 실행형 User Operator 배포 구성의 예

   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: strimzi-user-operator
     labels:
       app: strimzi
   spec:
     # ...
     template:
       # ...
       spec:
         # ...
         containers:
           - name: strimzi-user-operator
             # ...
             env:
               - name: STRIMZI_NAMESPACE 

   1

   
                 valueFrom:
                   fieldRef:
                     fieldPath: metadata.namespace
               - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS 

   2

   
                 value: my-kafka-bootstrap-address:9092
               - name: STRIMZI_CA_CERT_NAME 

   3

   
                 value: my-cluster-clients-ca-cert
               - name: STRIMZI_CA_KEY_NAME 

   4

   
                 value: my-cluster-clients-ca
               - name: STRIMZI_LABELS 

   5

   
                 value: "strimzi.io/cluster=my-cluster"
               - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS 

   6

   
                 value: "120000"
               - name: STRIMZI_WORK_QUEUE_SIZE 

   7

   
                 value: 10000
               - name: STRIMZI_CONTROLLER_THREAD_POOL_SIZE 

   8

   
                 value: 10
               - name: STRIMZI_USER_OPERATIONS_THREAD_POOL_SIZE 

   9

   
                 value: 4
               - name: STRIMZI_LOG_LEVEL 

   10

   
                 value: INFO
               - name: STRIMZI_GC_LOG_ENABLED 

   11

   
                 value: "true"
               - name: STRIMZI_CA_VALIDITY 

   12

   
                 value: "365"
               - name: STRIMZI_CA_RENEWAL 

   13

   
                 value: "30"
               - name: STRIMZI_JAVA_OPTS 

   14

   
                 value: "-Xmx=512M -Xms=256M"
               - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 

   15

   
                 value: "-Djavax.net.debug=verbose -DpropertyName=value"
               - name: STRIMZI_SECRET_PREFIX 

   16

   
                 value: "kafka-"
               - name: STRIMZI_ACLS_ADMIN_API_SUPPORTED 

   17

   
                 value: "true"
               - name: STRIMZI_MAINTENANCE_TIME_WINDOWS 

   18

   
                 value: '* * 8-10 * * ?;* * 14-15 * * ?'
               - name: STRIMZI_KAFKA_ADMIN_CLIENT_CONFIGURATION 

   19

   
                 value: |
                   default.api.timeout.ms=120000
                   request.timeout.ms=60000

   1

   User Operator가 KafkaUser 리소스를 조사하는 OpenShift 네임스페이스입니다. 하나의 네임스페이스만 지정할 수 있습니다.

   2

   Kafka 클러스터의 모든 브로커를 검색하고 연결할 부트스트랩 브로커 주소의 호스트 및 포트 쌍입니다. 서버가 다운된 경우 쉼표로 구분된 목록을 사용하여 두 개 또는 세 개의 브로커 주소를 지정합니다.

   3

   mTLS 인증을 위해 새 사용자 인증서에 서명하는 CA(인증 기관)의 공개 키(ca.crt) 값이 포함된 OpenShift Secret.

   4

   mTLS 인증을 위해 새 사용자 인증서에 서명하는 CA의 개인 키(ca.key) 값이 포함된 OpenShift 보안입니다.

   5

   User Operator에서 관리하는 KafkaUser 리소스를 식별하는 레이블입니다. 이는 Kafka 클러스터의 이름일 필요는 없습니다. KafkaUser 리소스에 할당된 레이블일 수 있습니다. User Operator를 두 개 이상 배포하는 경우 레이블은 각각 고유해야 합니다. 즉, 운영자는 동일한 리소스를 관리할 수 없습니다.

   6

   주기적 조정 간격(밀리초)입니다. 기본값은 120000 (2분)입니다.

   7

   컨트롤러 이벤트 큐의 크기입니다. 큐 크기는 User Operator가 작동할 것으로 예상되는 최대 사용자 수만큼 커야 합니다. 기본값은 1024 입니다.

   8

   사용자를 조정하기 위한 작업자 풀의 크기입니다. 더 큰 풀에는 더 많은 리소스가 필요할 수 있지만 더 많은 KafkaUser 리소스도 처리합니다. 기본값은 50 입니다.

   9

   Kafka Admin API 및 OpenShift 작업의 작업자 풀 크기입니다. 더 큰 풀에는 더 많은 리소스가 필요할 수 있지만 더 많은 KafkaUser 리소스도 처리합니다. 기본값은 4 입니다.

   10

   로깅 메시지를 출력하는 수준입니다. 수준을 ERROR,WARNING,INFO,DEBUG 또는 TRACE 로 설정할 수 있습니다.

   11

   가비지 컬렉션(GC) 로깅을 활성화합니다. 기본값은 true입니다.

   12

   CA의 유효 기간입니다. 기본값은 365 일입니다.

   13

   CA의 갱신 기간입니다. 갱신 기간은 현재 인증서의 만료 날짜로부터 역순으로 측정됩니다. 이전 인증서가 만료되기 전에 인증서 갱신을 시작하는 기본값은 30 일입니다.

   14

   (선택 사항) User Operator를 실행하는 JVM에서 사용하는 Java 옵션

   15

   (선택 사항) User Operator에 설정된 디버깅(-D) 옵션

   16

   (선택 사항) User Operator가 생성한 OpenShift 시크릿 이름에 대한 접두사입니다.

   17

   (선택 사항) Kafka 클러스터가 Kafka 관리자 API를 사용하여 권한 부여 ACL 규칙 관리를 지원하는지 여부를 나타냅니다. false 로 설정하면 User Operator에서 간단한 권한 부여 ACL 규칙이 있는 모든 리소스를 거부합니다. 이렇게 하면 Kafka 클러스터 로그에서 불필요한 예외를 방지할 수 있습니다. 기본값은 true입니다.

   18

   (선택 사항) 사용자 인증서가 갱신되는 유지 관리 시간을 정의하는 세미로 구분된 Cron Expression 목록입니다.

   19

   (선택 사항) User Operator에서 사용하는 Kafka 관리 클라이언트를 속성 형식으로 구성하는 구성 옵션입니다.

2. mTLS를 사용하여 Kafka 클러스터에 연결하는 경우 연결을 인증하는 데 사용되는 보안을 지정합니다. 그렇지 않으면 다음 단계로 이동합니다.

   mTLS 구성의 예

   # ....
   env:
     - name: STRIMZI_CLUSTER_CA_CERT_SECRET_NAME 

   1

   
       value: my-cluster-cluster-ca-cert
     - name: STRIMZI_EO_KEY_SECRET_NAME 

   2

   
       value: my-cluster-entity-operator-certs
   # ..."

   1

   Kafka 브로커 인증서에 서명하는 CA의 공개 키(ca.crt) 값이 포함된 OpenShift Secret 입니다.

   2

   Kafka 클러스터에 대한 mTLS 인증에 사용되는 인증서 공개 키(entity-operator.crt) 및 개인 키(entity-operator.key)가 포함된 OpenShift 시크릿 입니다.

3. User Operator를 배포합니다.

   oc create -f install/user-operator

4. 배포 상태를 확인합니다.

   oc get deployments

   출력에 배포 이름 및 준비 상태 표시

   NAME                   READY  UP-TO-DATE  AVAILABLE
   strimzi-user-operator  1/1    1           1

   READY 는 ready/expected 복제본 수를 표시합니다. AVAILABLE 출력에 1 이 표시되면 배포가 성공적으로 수행됩니다.

프로세스

1. Kafka 리소스의 구성에 플러그인 속성을 추가합니다.

   플러그인 속성은 이 예제 구성에 표시됩니다.

   Kafka 고정 할당량 플러그인 구성의 예

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       config:
         client.quota.callback.class: io.strimzi.kafka.quotas.StaticQuotaCallback 

   1

   
         client.quota.callback.static.produce: 1000000 

   2

   
         client.quota.callback.static.fetch: 1000000 

   3

   
         client.quota.callback.static.storage.soft: 400000000000 

   4

   
         client.quota.callback.static.storage.hard: 500000000000 

   5

   
         client.quota.callback.static.storage.check-interval: 5 

   6

   1

   Kafka 정적 할당량 플러그인을 로드합니다.

   2

   생산자 바이트 비율 임계값을 설정합니다. 이 예에서는 1Mbps입니다.

   3

   consumer byte-rate threshold를 설정합니다. 이 예에서는 1Mbps입니다.

   4

   스토리지에 대한 더 낮은 소프트 제한을 설정합니다. 이 예에서는 400GB입니다.

   5

   스토리지에 대해 더 높은 하드 제한을 설정합니다. 이 예에서는 500GB입니다.

   6

   스토리지의 점검 간격(초)을 설정합니다. 이 예에서는 5초입니다. 검사를 비활성화하려면 이 값을 0으로 설정할 수 있습니다.

2. 리소스를 업데이트합니다.

   oc apply -f <kafka_configuration_file>

프로세스

1. 삭제할 Pod 의 이름을 찾습니다.

   Kafka 브로커 Pod의 이름은 < cluster_name>-kafka-<index_number >로 지정됩니다. 여기서 < index_number >는 0에서 시작하여 총 복제본 수에서 1을 뺀 값입니다. 예를 들면 my-cluster-kafka-0 입니다.

2. oc annotate 를 사용하여 OpenShift에서 Pod 리소스에 주석을 답니다.

   oc annotate pod <cluster_name>-kafka-<index_number> strimzi.io/delete-pod-and-pvc="true"

3. 기본 영구 볼륨 클레임이 있는 주석이 달린 Pod가 삭제될 때 다음 조정을 기다린 후 다시 생성합니다.

프로세스

1. 삭제할 Pod 의 이름을 찾습니다.

   Zookeeper Pod의 이름은 < cluster_name>-zookeeper-<index_number >로 지정됩니다. 여기서 < index_number >는 0에서 시작하여 총 복제본 수에서 1을 뺀 값입니다. 예를 들면 my-cluster-zookeeper-0 입니다.

2. oc annotate 를 사용하여 OpenShift에서 Pod 리소스에 주석을 답니다.

   oc annotate pod <cluster_name>-zookeeper-<index_number> strimzi.io/delete-pod-and-pvc="true"

3. 기본 영구 볼륨 클레임이 있는 주석이 달린 Pod가 삭제될 때 다음 조정을 기다린 후 다시 생성합니다.