概要

Confluent for Kubernetes （CFK）は、プライベートクラウド環境（今回は Azure Kubernetes Service（AKS））に Confluent をデプロイして管理するためのクラウドネイティブのコントロールプレーンです。宣言型 API で Confluent Platform をカスタマイズ、デプロイ、管理するための標準的で簡素なインターフェイスが備わっています。

CFK を使用して AKS上にデプロイされている Confluent Platform に Connector を構成するためのワークフローの概要は以下のとおりです。

Kubernetes 環境を準備します（事前準備で完了）
Confluent for Kubernetes をデプロイします（事前準備で完了）
Confluent Platform をデプロイします（事前準備で完了）
Confluent Platform を Connector Plugin のコンテナで再デプロイします
Connector Plugin の確認

デプロイ後のイメージは以下となります。

ローカル環境

macOS Monterey 12.3.1
python 3.8.12
Azure CLI 2.34.1
helm v3.6.3
kubectl v1.21.3

事前準備

この記事を実行して、AKSクラスタ環境が構築されていること
この記事を実行して、AKS 上に Confluent Platform をデプロイされていること
この記事を参照して、新たに定義した Connector Plugin がはいっている Connect コンテナイメージが DockerHub に公開されていること

Confluent Platform を Connector Plugin のコンテナで再デプロイ

Confluent Platform の構成変更

上記の事前準備「2.」の中の「confluent_platform_ccc.yaml」を CosmosDB の Connector Plugin が入っている Docker イメージを利用するように変更します

confluent_platform_ccc.yaml


##以下の部分の変更
apiVersion: platform.confluent.io/v1beta1
kind: Connect
metadata:
  name: connect
  namespace: akscfk231
spec:
  replicas: 1
  image:
    application: keisz/confluent_connector:1.0         # <--- ここを変更
    init: confluentinc/confluent-init-container:2.3.0
  dependencies:
    kafka:
      bootstrapEndpoint: kafka:9071

Confluent Platform の再デプロイ

$ kubectl apply -f confluent_platform_ccc.yaml

zookeeper.platform.confluent.io/zookeeper unchanged
kafka.platform.confluent.io/kafka unchanged
connect.platform.confluent.io/connect configured
ksqldb.platform.confluent.io/ksqldb unchanged
controlcenter.platform.confluent.io/controlcenter unchanged
schemaregistry.platform.confluent.io/schemaregistry unchanged
kafkarestproxy.platform.confluent.io/kafkarestproxy unchanged

Connector Plugin の確認

Pod/ｓｖｃの変更

$ kubectl get pods

NAME                                  READY   STATUS    RESTARTS   AGE
confluent-operator-76d7677b8c-pdp8s   1/1     Running   0          13m
connect-0                             1/1     Running   0          111s
controlcenter-0                       1/1     Running   0          8m20s
kafka-0                               1/1     Running   0          9m51s
kafka-1                               1/1     Running   0          9m51s
kafka-2                               1/1     Running   0          9m51s
kafkarestproxy-0                      1/1     Running   0          8m21s
ksqldb-0                              1/1     Running   0          8m20s
schemaregistry-0                      1/1     Running   0          8m20s
zookeeper-0                           1/1     Running   0          12m
zookeeper-1                           1/1     Running   0          12m
zookeeper-2                           1/1     Running   0          12m


$ kubectl get svc

NAME                        TYPE        CLUSTER-IP   EXTERNAL-IP   PORT(S)                                                          AGE
confluent-operator          ClusterIP   10.1.0.231   <none>        7778/TCP                                                         15m
connect                     ClusterIP   None         <none>        8083/TCP,7203/TCP,7777/TCP,7778/TCP                              14m
connect-0-internal          ClusterIP   10.1.0.232   <none>        8083/TCP,7203/TCP,7777/TCP,7778/TCP                              14m
controlcenter               ClusterIP   None         <none>        9021/TCP,7203/TCP,7777/TCP,7778/TCP                              10m
controlcenter-0-internal    ClusterIP   10.1.0.95    <none>        9021/TCP,7203/TCP,7777/TCP,7778/TCP                              10m
kafka                       ClusterIP   None         <none>        9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   11m
kafka-0-internal            ClusterIP   10.1.0.181   <none>        9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   11m
kafka-1-internal            ClusterIP   10.1.0.103   <none>        9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   11m
kafka-2-internal            ClusterIP   10.1.0.27    <none>        9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   11m
kafkarestproxy              ClusterIP   None         <none>        8082/TCP,7203/TCP,7777/TCP,7778/TCP                              10m
kafkarestproxy-0-internal   ClusterIP   10.1.0.107   <none>        8082/TCP,7203/TCP,7777/TCP,7778/TCP                              10m
ksqldb                      ClusterIP   None         <none>        8088/TCP,7203/TCP,7777/TCP,7778/TCP                              10m
ksqldb-0-internal           ClusterIP   10.1.0.22    <none>        8088/TCP,7203/TCP,7777/TCP,7778/TCP                              10m
schemaregistry              ClusterIP   None         <none>        8081/TCP,7203/TCP,7777/TCP,7778/TCP                              10m
schemaregistry-0-internal   ClusterIP   10.1.0.227   <none>        8081/TCP,7203/TCP,7777/TCP,7778/TCP                              10m
zookeeper                   ClusterIP   None         <none>        2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP            14m
zookeeper-0-internal        ClusterIP   10.1.0.66    <none>        2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP            14m
zookeeper-1-internal        ClusterIP   10.1.0.21    <none>        2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP            14m
zookeeper-2-internal        ClusterIP   10.1.0.118   <none>        2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP            14m

Connector の確認

確認するために Connect をローカルホストにリダイレクトします

$ kubectl port-forward --address localhost svc/connect 8083:8083
Forwarding from 127.0.0.1:8083 -> 8083
Forwarding from [::1]:8083 -> 8083

※ CTRL+C で終了できます

別のターミナルを立ち上げ、Connector の Plug-in が存在することを確認する（CosmosDB のPluginの確認）

$ curl http://localhost:8083/connector-plugins | jq

  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100   504  100   504    0     0   2030      0 --:--:-- --:--:-- --:--:--  2032
[
  {
    "class": "com.azure.cosmos.kafka.connect.sink.CosmosDBSinkConnector",
    "type": "sink",
    "version": "null"
  },
  {
    "class": "com.azure.cosmos.kafka.connect.source.CosmosDBSourceConnector",
    "type": "source",
    "version": "null"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorCheckpointConnector",
    "type": "source",
    "version": "1"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorHeartbeatConnector",
    "type": "source",
    "version": "1"
  },
  {
    "class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
    "type": "source",
    "version": "1"
  }
]

Confluent Control Center 上での確認

$ kubectl confluent dashboard controlcenter
http://localhost:9021

ブラウザが自動的に立ち上がり、Confluent Contorol Center にログインし、「Connect」を確認します

後処理

Pod / secret / namespace のアンインストール方法

## Pod : confluent-operator
$ helm delete confluent-operator

## Pod : confluent-platform
$ kubectl delete -f confluent_platform_ccc.yaml

## namespace の削除方法（namespace配下のPodは全て削除される）
$ kubectl delete namespace akscfk231

AKSクラスターの停止・起動

$ az aks stop -g rg_ituru_aks01 -n aks_ituru_cp01
$ az aks start -g rg_ituru_aks01 -n aks_ituru_cp01

Check for any error messages in events

kubectl get events

まとめ

Connector Plugin がインストールされている（自作構成） Dockerイメージで、Confluent Platform が構成できることを確認しました。

参考情報

以下の情報を参考にさせていただきました

Confluent for Kubernetes