5.4. ブローカー

検証

1. kn コマンドを使用して、既存のブローカーを一覧表示します。

   $ kn broker list

   出力例

   NAME      URL                                                                     AGE   CONDITIONS   READY   REASON
   default   http://broker-ingress.knative-eventing.svc.cluster.local/test/default   45s   5 OK / 5     True

2. オプション: OpenShift Container Platform Web コンソールを使用している場合、Developer パースペクティブの Topology ビューに移動し、ブローカーが存在することを確認できます。

   

手順

1. Trigger オブジェクトを、eventing.knative.dev/injection: enabled アノテーションを付けて YAML ファイルとして作成します。

   apiVersion: eventing.knative.dev/v1
   kind: Trigger
   metadata:
     annotations:
       eventing.knative.dev/injection: enabled
     name: <trigger_name>
   spec:
     broker: default
     subscriber: 1
       ref:
         apiVersion: serving.knative.dev/v1
         kind: Service
         name: <service_name>

   1

   トリガーがイベントを送信するイベントシンクまたは サブスクライバー の詳細を指定します。

2. Trigger YAML ファイルを適用します。

   $ oc apply -f <filename>

1. 以下の oc コマンドを入力してブローカーを取得します。

   $ oc -n <namespace> get broker default

   出力例

   NAME      READY     REASON    URL                                                                     AGE
   default   True                http://broker-ingress.knative-eventing.svc.cluster.local/test/default   3m56s

2. オプション: OpenShift Container Platform Web コンソールを使用している場合、Developer パースペクティブの Topology ビューに移動し、ブローカーが存在することを確認できます。

   

1. oc コマンドを使用してブローカーを取得します。

   $ oc -n <namespace> get broker <broker_name>

   コマンドの例

   $ oc -n default get broker default

   出力例

   NAME      READY     REASON    URL                                                                     AGE
   default   True                http://broker-ingress.knative-eventing.svc.cluster.local/test/default   3m56s

2. オプション: OpenShift Container Platform Web コンソールを使用している場合、Developer パースペクティブの Topology ビューに移動し、ブローカーが存在することを確認できます。

   

手順

1. eventing.knative.dev/injection=enabled ラベルを namespace から削除します。

   $ oc label namespace <namespace> eventing.knative.dev/injection-

   アノテーションを削除すると、Knative では削除後にブローカーを再作成できなくなります。

2. 選択された namespace からブローカーを削除します。

   $ oc -n <namespace> delete broker <broker_name>

手順

1. Developer パースペクティブで、+Add Broker に移動します。Broker ページが表示されます。
2. オプション:ブローカーの Name を更新します。名前を更新しない場合、生成されたブローカーの名前は default になります。
3. Create をクリックします。

1. Developer パースペクティブで、Topology に移動します。

2. mt-broker-ingress、mt-broker-filter、および mt-broker-controller コンポーネントを表示します。

   

手順

1. OpenShift Container Platform Web コンソールの Administrator パースペクティブで、 Serverless Eventing に移動します。
2. Create 一覧で、Broker を選択します。Create Broker ページに移動します。
3. オプション: ブローカーの YAML 設定を変更します。
4. Create をクリックします。

手順

1. kafka-broker-config ConfigMap を変更するか、以下の設定が含まれる独自の ConfigMap を作成します。

   apiVersion: v1
   kind: ConfigMap
   metadata:
     name: <config_map_name> 1
     namespace: <namespace> 2
   data:
     default.topic.partitions: <integer> 3
     default.topic.replication.factor: <integer> 4
     bootstrap.servers: <list_of_servers> 5

   1

   ConfigMap 名。

   2

   ConfigMap が存在する namespace。

   3

   Kafka ブローカーのトピックパーティションの数。これは、イベントをブローカーに送信する速度を制御します。パーティションが多い場合には、コンピュートリソースが多く必要です。

   4

   トピックメッセージのレプリケーション係数。これにより、データ損失を防ぐことができます。レプリケーション係数を増やすには、より多くのコンピュートリソースとストレージが必要になります。

   5

   ブートストラップサーバーのコンマ区切りリスト。これは、OpenShift Container Platform クラスターの内部または外部にある可能性があり、ブローカーがイベントを受信してイベントを送信する Kafka クラスターのリストです。

   重要

   default.topic.replication.factor の値は、クラスター内の Kafka ブローカーインスタンスの数以下である必要があります。たとえば、Kafka ブローカーが 1 つしかない場合には、default.topic.replication.factor の値は "1" を超える値にすることはできません。

   Kafka ブローカーの ConfigMap の例

   apiVersion: v1
   kind: ConfigMap
   metadata:
     name: kafka-broker-config
     namespace: knative-eventing
   data:
     default.topic.partitions: "10"
     default.topic.replication.factor: "3"
     bootstrap.servers: "my-cluster-kafka-bootstrap.kafka:9092"

2. ConfigMap を適用します。

   $ oc apply -f <config_map_filename>

3. Kafka Broker オブジェクトの ConfigMap を指定します。

   Broker オブジェクトの例

   apiVersion: eventing.knative.dev/v1
   kind: Broker
   metadata:
     name: <broker_name> 1
     namespace: <namespace> 2
     annotations:
       eventing.knative.dev/broker.class: Kafka 3
   spec:
     config:
       apiVersion: v1
       kind: ConfigMap
       name: <config_map_name> 4
       namespace: <namespace> 5
   ...

   1

   ブローカー名。

   2

   ブローカーが存在する namespace。

   3

   ブローカークラスアノテーション。この例では、ブローカーはクラス値 Kafka を使用する Kafka ブローカーです。

   4

   ConfigMap 名。

   5

   ConfigMap が存在する namespace。

4. ブローカーを適用します。

   $ oc apply -f <broker_filename>

手順

1. Trigger オブジェクトを作成または変更し、kafka.eventing.knative.dev/delivery.order アノテーションを設定します。

   apiVersion: eventing.knative.dev/v1
   kind: Trigger
   metadata:
     name: <trigger_name>
     annotations:
        kafka.eventing.knative.dev/delivery.order: ordered
   ...

   サポートされているコンシューマー配信保証は次のとおりです。

   unordered

   順序付けられていないコンシューマーは、適切なオフセット管理を維持しながら、メッセージを順序付けずに配信するノンブロッキングコンシューマーです。

   ordered

   順序付きコンシューマーは、CloudEvent サブスクライバーからの正常な応答を待ってから、パーティションの次のメッセージを配信する、パーティションごとのブロックコンシューマーです。

   デフォルトの順序保証は unordered です。

2. Trigger オブジェクトを適用します。

   $ oc apply -f <filename>