1.环境准备

使用pulsar官方helm仓库 https://pulsar.apache.org/charts 部署pulsar集群

2.添加kop插件

Apache官方镜像是不带kop插件的，需要使用streamnative提供的已默认集成kop插件的pulsar镜像（当然也可以在Apache官方镜像基础上添加kop插件自己制作镜像），可以到dockerhub上自行查找所需要的镜像版本，这里我使用的镜像是 streamnative/sn-pulsar:2.9.2.24

3.添加broker kop配置

找到broker的configMap，添加以下配置

  PULSAR_PREFIX_brokerEntryMetadataInterceptors: org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor
  PULSAR_PREFIX_kafkaListeners: 'PLAINTEXT://0.0.0.0:9092'
  PULSAR_PREFIX_messagingProtocols: kafka
  allowAutoTopicCreationType: partitioned
  brokerDeleteInactiveTopicsEnabled: 'false'

下面还有一步比较关键的配置，需要在broker的有状态副本集中增加环境变量，配置kafka broker对外广播的地址，需要引入k8s当前pod名称的环境变量，如下

          env:
            - name: MY_POD_NAME
              valueFrom:
                fieldRef:
                  apiVersion: v1
                  fieldPath: metadata.name
            - name: PULSAR_PREFIX_kafkaAdvertisedListeners
              value: >-
                PLAINTEXT://$(MY_POD_NAME).pulsar-broker.xxx.svc.cluster.local:19092

当然还要在port下增加kafka的端口，hostPort可以用于k8s外访问

            - name: kafka
              hostPort: 19092
              containerPort: 9092
              protocol: TCP

k8s外访问只需要配置一下本地hosts即可

172.18.100.33 ks-pulsar-broker-0.pulsar-broker.xxx.svc.cluster.local
172.18.100.35 ks-pulsar-broker-1.pulsar-broker.xxx.svc.cluster.local
172.18.100.31 ks-pulsar-broker-2.pulsar-broker.xxx.svc.cluster.local