2.5. Apache Kafka의 소스

절차

1. 개발자 화면에서 +추가 페이지로 이동하여 이벤트 소스를 선택합니다.
2. 이벤트 소스 페이지의 유형 섹션에서 Kafka 소스를 선택합니다.

3. Kafka 소스 설정을 구성합니다.

   1. 쉼표로 구분된 부트스트랩 서버 목록을 추가합니다.
   2. 쉼표로 구분된 주제 목록을 추가합니다.
   3. 소비자 그룹을 추가합니다.
   4. 생성한 서비스 계정의 서비스 계정 이름을 선택합니다.
   5. 이벤트 소스로 싱크를 선택합니다. 싱크는 채널, 브로커 또는 서비스와 같은 리소스이거나 URI일 수 있습니다.
   6. Kafka 이벤트 소스로 이름을 입력합니다.
4. 생성을 클릭합니다.

1. 개발자 화면에서 토폴로지로 이동합니다.

2. Kafka 이벤트 소스 및 싱크를 확인합니다.

   

절차

1. Kafka 이벤트 소스가 작동하는지 확인하려면 수신 이벤트를 서비스 로그에 덤프하는 Knative 서비스를 생성합니다.

   $ kn service create event-display \
       --image quay.io/openshift-knative/knative-eventing-sources-event-display

2. KafkaSource CR을 생성합니다.

   $ kn source kafka create <kafka_source_name> \
       --servers <cluster_kafka_bootstrap>.kafka.svc:9092 \
       --topics <topic_name> --consumergroup my-consumer-group \
       --sink event-display

   참고

   이 명령의 자리 표시자 값을 소스 이름, 부트스트랩 서버 및 주제의 값으로 바꿉니다.

   --servers, --topics, --consumergroup 옵션은 Kafka 클러스터에 대한 연결 매개 변수를 지정합니다. --consumergroup 옵션은 선택 사항입니다.

3. 선택 사항: 생성한 KafkaSource CR에 대한 세부 정보를 확인합니다.

   $ kn source kafka describe <kafka_source_name>

   출력 예

   Name:              example-kafka-source
   Namespace:         kafka
   Age:               1h
   BootstrapServers:  example-cluster-kafka-bootstrap.kafka.svc:9092
   Topics:            example-topic
   ConsumerGroup:     example-consumer-group
   
   Sink:
     Name:       event-display
     Namespace:  default
     Resource:   Service (serving.knative.dev/v1)
   
   Conditions:
     OK TYPE            AGE REASON
     ++ Ready            1h
     ++ Deployed         1h
     ++ SinkProvided     1h

검증 단계

1. Kafka 인스턴스를 트리거하여 메시지를 항목에 보냅니다.

   $ oc -n kafka run kafka-producer \
       -ti --image=quay.io/strimzi/kafka:latest-kafka-2.7.0 --rm=true \
       --restart=Never -- bin/kafka-console-producer.sh \
       --broker-list <cluster_kafka_bootstrap>:9092 --topic my-topic

   프롬프트에 메시지를 입력합니다. 이 명령은 다음을 가정합니다.

   • Kafka 클러스터는 kafka 네임스페이스에 설치됩니다.
   • my-topic 주제를 사용하도록 KafkaSource 오브젝트가 구성되어 있습니다.

2. 로그를 보고 메시지가 도착했는지 확인합니다.

   $ oc logs $(oc get pod -o name | grep event-display) -c user-container

   출력 예

   ☁️  cloudevents.Event
   Validation: valid
   Context Attributes,
     specversion: 1.0
     type: dev.knative.kafka.event
     source: /apis/v1/namespaces/default/kafkasources/example-kafka-source#example-topic
     subject: partition:46#0
     id: partition:46/offset:0
     time: 2021-03-10T11:21:49.4Z
   Extensions,
     traceparent: 00-161ff3815727d8755848ec01c866d1cd-7ff3916c44334678-00
   Data,
     Hello!

절차

1. KafkaSource 오브젝트를 YAML 파일로 생성합니다.

   apiVersion: sources.knative.dev/v1beta1
   kind: KafkaSource
   metadata:
     name: <source_name>
   spec:
     consumerGroup: <group_name> 1
     bootstrapServers:
     - <list_of_bootstrap_servers>
     topics:
     - <list_of_topics> 2
     sink:
     - <list_of_sinks> 3

   1

   소비자 그룹은 동일한 그룹 ID를 사용하고 한 주제의 데이터를 사용하는 소비자 그룹입니다.

   2

   주제에서는 데이터 스토리지의 대상을 제공합니다. 각 주제는 하나 이상의 파티션으로 나뉩니다.

   3

   싱크는 소스에서 이벤트를 보내는 위치를 지정합니다.

   중요

   OpenShift Serverless의 KafkaSource 개체에 대한 API의 v1beta1 버전만 지원됩니다. 이 버전은 더 이상 사용되지 않으므로 이 API의 v1alpha1 버전을 사용하지 마십시오.

   KafkaSource 오브젝트의 예

   apiVersion: sources.knative.dev/v1beta1
   kind: KafkaSource
   metadata:
     name: kafka-source
   spec:
     consumerGroup: knative-group
     bootstrapServers:
     - my-cluster-kafka-bootstrap.kafka:9092
     topics:
     - knative-demo-topic
     sink:
       ref:
         apiVersion: serving.knative.dev/v1
         kind: Service
         name: event-display

2. KafkaSource YAML 파일을 적용합니다.

   $ oc apply -f <filename>

절차

1. 선택한 네임스페이스에서 인증서 파일을 시크릿으로 생성합니다.

   $ oc create secret -n <namespace> generic <kafka_auth_secret> \
     --from-file=ca.crt=caroot.pem \
     --from-literal=password="SecretPassword" \
     --from-literal=saslType="SCRAM-SHA-512" \ 1
     --from-literal=user="my-sasl-user"

   1

   SASL 유형은 PLAIN,SCRAM-SHA-256 또는 SCRAM-SHA-512 일 수 있습니다.

2. 다음 사양 구성을 포함하도록 Kafka 소스를 생성하거나 수정합니다.

   apiVersion: sources.knative.dev/v1beta1
   kind: KafkaSource
   metadata:
     name: example-source
   spec:
   ...
     net:
       sasl:
         enable: true
         user:
           secretKeyRef:
             name: <kafka_auth_secret>
             key: user
         password:
           secretKeyRef:
             name: <kafka_auth_secret>
             key: password
         type:
           secretKeyRef:
             name: <kafka_auth_secret>
             key: saslType
       tls:
         enable: true
         caCert: 1
           secretKeyRef:
             name: <kafka_auth_secret>
             key: ca.crt
   ...

   1

   퍼블릭 클라우드 Kafka 서비스를 사용하는 경우 caCert 사양이 필요하지 않습니다.