AMQ Streams Kafka 브리지 사용

프로세스

1. 아직 수행하지 않은 경우 Kafka 브리지 설치 아카이브의 압축을 임의의 디렉터리에 풉니다.

2. 구성 속성을 매개변수로 사용하여 Kafka Bridge 스크립트를 실행합니다.

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

   ./bin/kafka_bridge_run.sh --config-file=<path>/application.properties

   Copy to Clipboard Toggle word wrap

3. 로그에 설치에 성공했는지 확인합니다.

   HTTP-Kafka Bridge started and listening on port 8080
   HTTP-Kafka Bridge bootstrap servers localhost:9092

   Copy to Clipboard Toggle word wrap

프로세스

1. Kafka 브리지를 사용하여 생성한 항목에 세 개의 메시지를 생성합니다.

   curl -X POST \
     http://localhost:8080/topics/bridge-quickstart-topic \
     -H 'content-type: application/vnd.kafka.json.v2+json' \
     -d '{
       "records": [
           {
               "key": "my-key",
               "value": "sales-lead-0001"
           },
           {
               "value": "sales-lead-0002",
               "partition": 2
           },
           {
               "value": "sales-lead-0003"
           }
       ]
   }'

   Copy to Clipboard Toggle word wrap
   • sales-lead-0001 은 키의 해시를 기반으로 파티션으로 전송됩니다.
   • sales-lead-0002 는 파티션 2로 직접 전송됩니다.
   • sales-lead-0003 은 라운드 로빈 방법을 사용하여 bridge-quickstart-topic 주제의 파티션으로 전송됩니다.

2. 요청이 성공하면 Kafka 브리지는 200 코드 및 application/vnd.kafka.v2+json 의 content-type 헤더와 함께 오프셋 배열을 반환합니다. 각 메시지에 대해 오프셋 배열은 다음을 설명합니다.

   • 메시지가 전송된 파티션

   • 파티션의 현재 메시지 오프셋

     응답 예

     #...
     {
       "offsets":[
         {
           "partition":0,
           "offset":0
         },
         {
           "partition":2,
           "offset":0
         },
         {
           "partition":0,
           "offset":1
         }
       ]
     }

     Copy to Clipboard Toggle word wrap

프로세스

1. bridge-quickstart-consumer-group 이라는 새 소비자 그룹에 Kafka 브리지 소비자를 생성합니다.

   curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group \
     -H 'content-type: application/vnd.kafka.v2+json' \
     -d '{
       "name": "bridge-quickstart-consumer",
       "auto.offset.reset": "earliest",
       "format": "json",
       "enable.auto.commit": false,
       "fetch.min.bytes": 512,
       "consumer.request.timeout.ms": 30000
     }'

   Copy to Clipboard Toggle word wrap
   • 소비자의 이름은 bridge-quickstart-consumer 로 지정되며 포함된 데이터 형식은 json 으로 설정됩니다.
   • 일부 기본 구성 설정이 정의됩니다.

   • enable.auto.commit 설정은 false 이므로 소비자는 로그에 오프셋을 자동으로 커밋하지 않습니다. 이 빠른 시작의 뒷부분에서 오프셋을 수동으로 커밋합니다.

     요청이 성공하면 Kafka 브리지는 응답 본문의 소비자 ID(instance_id) 및 기본 URL(base_uri)을 200 코드와 함께 반환합니다.

     응답 예

     #...
     {
       "instance_id": "bridge-quickstart-consumer",
       "base_uri":"http://<bridge_id>-bridge-service:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer"
     }

     Copy to Clipboard Toggle word wrap

2. 이 빠른 시작의 다른 소비자 작업에 사용할 기본 URL(base_uri)을 복사합니다.

프로세스

1. 주제 및 파티션에 대한 메시지 전달에 설명된 대로 Kafka Bridge 소비자에 추가 메시지를 생성합니다.

2. GET 요청을 레코드 끝점에 제출합니다.

   curl -X GET http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/records \
     -H 'accept: application/vnd.kafka.json.v2+json'

   Copy to Clipboard Toggle word wrap

   Kafka Bridge 소비자를 생성하고 구독하면 첫 번째 GET 요청이 비어 있는 응답을 반환합니다. 폴링 작업에서 파티션을 할당하는 재조정 프로세스를 시작하기 때문입니다.

3. 두 단계를 반복하여 Kafka 브리지 소비자에서 메시지를 검색합니다.

   Kafka Bridge는 200 코드와 함께 응답 본문에 있는 주제 이름, 키, 값, 파티션 및 오프셋 Cryostat-지정을 설명하는 messages의 배열을 반환합니다. 메시지는 기본적으로 최신 오프셋에서 검색됩니다.

   HTTP/1.1 200 OK
   content-type: application/vnd.kafka.json.v2+json
   #...
   [
     {
       "topic":"bridge-quickstart-topic",
       "key":"my-key",
       "value":"sales-lead-0001",
       "partition":0,
       "offset":0
     },
     {
       "topic":"bridge-quickstart-topic",
       "key":null,
       "value":"sales-lead-0003",
       "partition":0,
       "offset":1
     },
   #...

   Copy to Clipboard Toggle word wrap
   참고

   빈 응답이 반환되면 주제 및 파티션에 대한 메시지 전달에 설명된 대로 소비자에 더 많은 레코드를 생성한 다음 메시지를 다시 검색해 보십시오.

프로세스

1. quickstart-bridge-topic 주제의 파티션 0에 대한 특정 오프셋을 찾습니다.

   curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/positions \
     -H 'content-type: application/vnd.kafka.v2+json' \
     -d '{
       "offsets": [
           {
               "topic": "bridge-quickstart-topic",
               "partition": 0,
               "offset": 2
           }
       ]
   }'

   Copy to Clipboard Toggle word wrap

   요청이 성공하면 Kafka 브리지는 204 코드만 반환합니다.

2. GET 요청을 레코드 끝점에 제출합니다.

   curl -X GET http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/records \
     -H 'accept: application/vnd.kafka.json.v2+json'

   Copy to Clipboard Toggle word wrap

   Kafka 브리지는 원하는 오프셋에서 메시지를 반환합니다.

3. 동일한 파티션의 마지막 오프셋을 찾아 기본 메시지 검색 동작을 복원합니다. 이번에는 위치/엔드 엔드포인트를 사용합니다.

   curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/positions/end \
     -H 'content-type: application/vnd.kafka.v2+json' \
     -d '{
       "partitions": [
           {
               "topic": "bridge-quickstart-topic",
               "partition": 0
           }
       ]
   }'

   Copy to Clipboard Toggle word wrap

   요청이 성공하면 Kafka 브리지는 또 다른 204 코드를 반환합니다.

프로세스

1. Kafka Bridge 설치 아카이브와 함께 제공된 application.properties 파일을 편집합니다.

   속성 파일을 사용하여 Kafka 및 HTTP 관련 속성을 지정합니다.

   1. Kafka 소비자 및 생산자와 관련된 속성을 포함하여 표준 Kafka 관련 속성을 구성합니다.

      다음을 사용하십시오.

      • Kafka 클러스터에 대한 호스트/포트 연결을 정의하는 Kafka .bootstrap.servers
      • Kafka.producer.acks HTTP 클라이언트에 승인을 제공

      • Kafka.consumer.auto.offset.reset 에서 오프셋의 재설정을 관리하는 방법

        Kafka 속성 구성에 대한 자세한 내용은 Apache Kafka 웹 사이트를참조하십시오.

   2. Kafka 클러스터에 대한 HTTP 액세스를 활성화하도록 HTTP 관련 속성을 구성합니다.

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

      bridge.id=my-bridge
      http.host=0.0.0.0
      http.port=8080 

      1

      
      http.cors.enabled=true 

      2

      
      http.cors.allowedOrigins=https://strimzi.io 

      3

      
      http.cors.allowedMethods=GET,POST,PUT,DELETE,OPTIONS,PATCH 

      4

      Copy to Clipboard Toggle word wrap

      1

      포트 8080에서 수신 대기할 Kafka 브리지의 기본 HTTP 구성입니다.

      2

      CORS를 활성화하려면 true 로 설정합니다.

      3

      쉼표로 구분된 허용된 CORS 원본 목록입니다. URL 또는 Java 정규식을 사용할 수 있습니다.

      4

      CORS에 허용되는 HTTP 메서드의 쉼표로 구분된 목록입니다.

2. 구성 파일을 저장합니다.

프로세스

1. Kafka Bridge 설치 아카이브와 함께 제공된 application.properties 파일을 편집합니다.

   bridge.tracing 속성을 사용하여 사용하려는 추적을 활성화합니다.

   OpenTelemetry를 활성화하는 구성 예

   bridge.tracing=opentelemetry 

   1

   Copy to Clipboard Toggle word wrap

   1

   OpenTelemetry를 활성화하는 속성은 줄 시작 시 # 을 제거하여 주석 처리를 해제합니다.

   추적이 활성화된 경우 Kafka Bridge 스크립트를 실행할 때 추적을 초기화합니다.

2. 구성 파일을 저장합니다.

3. 추적의 환경 변수를 설정합니다.

   OpenTelemetry의 환경 변수

   OTEL_SERVICE_NAME=my-tracing-service 

   1

   
   OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 

   2

   Copy to Clipboard Toggle word wrap

   1

   OpenTelemetry 추적기 서비스의 이름입니다.

   2

   포트 4317에서 범위를 수신 대기하는 gRPC 기반 OTLP 끝점입니다.

4. 추적을 위해 활성화된 속성을 사용하여 Kafka Bridge 스크립트를 실행합니다.

   OpenTelemetry를 사용하여 Kafka 브리지 실행

   ./bin/kafka_bridge_run.sh --config-file=<path>/application.properties

   Copy to Clipboard Toggle word wrap

   Kafka 브리지의 내부 소비자 및 생산자가 이제 추적에 사용할 수 있습니다.