bookkeeper.yaml

---
# SSDs for bookie journal storage
kind: StorageClass
apiVersion: storage.k8s.io/v1
metadata:
  namespace: pulsar
  name: bookie-ssd
provisioner: kubernetes.io/aws-ebs
parameters:
  type: gp2
  fsType: xfs
#  To create encrypted ebs volume using kms
#  encrypted: "true"
#  kmsKeyId: <enter the key id here>
reclaimPolicy: Retain
allowVolumeExpansion: true
volumeBindingMode: WaitForFirstConsumer

---
# HDDs for bookie ledger storage
kind: StorageClass
apiVersion: storage.k8s.io/v1
metadata:
  namespace: pulsar
  name: bookie-hdd
provisioner: kubernetes.io/aws-ebs
parameters:
  type: gp2
  fsType: xfs
#  To create encrypted ebs volume using kms
#  encrypted: "true"
#  kmsKeyId: <enter the key id here>
reclaimPolicy: Retain
allowVolumeExpansion: true
volumeBindingMode: WaitForFirstConsumer

---
apiVersion: v1
kind: ConfigMap
metadata:
  namespace: pulsar
  name: bookie-config
data:
  BOOKIE_MEM: '"-Dio.netty.leakDetectionLevel=disabled -Dio.netty.recycler.linkCapacity=1024 -XX:+ParallelRefProcEnabled -XX:+UnlockExperimentalVMOptions -XX:+AggressiveOpts -XX:+DoEscapeAnalysis -XX:ParallelGCThreads=32 -XX:ConcGCThreads=32 -XX:G1NewSizePercent=50 -XX:+DisableExplicitGC -XX:-ResizePLAB -XX:+ExitOnOutOfMemoryError -XX:+PerfDisableSharedMem -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintGCApplicationStoppedTime -XX:+PrintHeapAtGC -verbosegc -XX:G1LogLevel=finest -Xms512m -Xmx512m -XX:MaxDirectMemorySize=512m"'
  BOOKIE_GC: '"-XX:+UseG1GC -XX:MaxGCPauseMillis=10"'
  dbStorage_writeCacheMaxSizeMb: "2048" # Write cache size (direct memory)
  dbStorage_readAheadCacheMaxSizeMb: "2048" # Read cache size (direct memory)
  dbStorage_rocksDB_blockCacheSize: "4294967296"
  journalMaxSizeMB: "2048"
  zkServers: zk-0.zookeeper,zk-1.zookeeper,zk-2.zookeeper
  statsProviderClass: org.apache.bookkeeper.stats.prometheus.PrometheusMetricsProvider
  useHostNameAsBookieID: "true"
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  namespace: pulsar
  name: bookkeeper
  labels:
    app: pulsar
    component: bookkeeper
    role: bookie
spec:
  serviceName: bookkeeper
  selector:
      matchLabels:
          app: pulsar
          component: bookkeeper
          cluster: eu-central-1
          role: bookie
  replicas: 2
  template:
    metadata:
      labels:
        app: pulsar
        component: bookkeeper
        # Specify cluster to allow aggregation by cluster in
        # the metrics
        cluster: eu-central-1
        role: bookie
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "8000"

    spec:
      containers:
        - name: bookie
          image: apachepulsar/pulsar:latest
          command: ["sh", "-c"]
          args:
            - >
              bin/apply-config-from-env.py conf/bookkeeper.conf &&
              bin/apply-config-from-env.py conf/pulsar_env.sh &&
              bin/pulsar bookie
          ports:
            - containerPort: 3181
              name: client
          envFrom:
            - configMapRef:
                name: bookie-config
          env:

          volumeMounts:
            - name: journal-disk
              mountPath: /pulsar/data/bookkeeper/journal
            - name: ledgers-disk
              mountPath: /pulsar/data/bookkeeper/ledgers

      initContainers:
        # The first time, initialize BK metadata in zookeeper
        # Otherwise ignore error if it's already there
        - name: bookie-metaformat
          image: apachepulsar/pulsar:latest
          command: ["sh", "-c"]
          args:
            - >
              bin/apply-config-from-env.py conf/bookkeeper.conf &&
              bin/bookkeeper shell metaformat --nonInteractive || true;
          envFrom:
            - configMapRef:
                name: bookie-config

  # Uncomment the following if bookie should be deployed on a specific instance type
  #            nodeSelector:
  #              beta.kubernetes.io/instance-type: m4.2xlarge

  volumeClaimTemplates:
    - metadata:
        name: journal-disk
      spec:
        accessModes: ["ReadWriteOnce"]
        resources:
          requests:
            storage: 30Gi
        storageClassName: bookie-ssd
    - metadata:
        name: ledgers-disk
      spec:
        accessModes: ["ReadWriteOnce"]
        resources:
          requests:
            storage: 100Gi
        storageClassName: bookie-hdd

---
##
## Define the Bookie headless service
## In practice, in this case, it is only useful to have a view of
## all the bookie pods that are present
##
apiVersion: v1
kind: Service
metadata:
  namespace: pulsar
  name: bookkeeper
  labels:
    app: pulsar
    component: bookkeeper
    role: bookie
spec:
  ports:
    - port: 3181
      name: server
  clusterIP: None
  selector:
    app: pulsar
    component: bookkeeper

---
##
## Run BookKeeper auto-recovery from a different set of containers
## Auto-Recovery makes sure to restore the replication factor when any bookie
## crashes and it's not recovering on its own.
##
apiVersion: apps/v1
kind: Deployment
metadata:
  namespace: pulsar
  name: bookie-autorecovery
spec:
  selector:
      matchLabels:
          app: pulsar
          component: bookkeeper-replication
          role: bookie
  replicas: 2
  template:
    metadata:
      labels:
        app: pulsar
        component: bookkeeper-replication
        role: bookie
    spec:
      containers:
        - name: replication-worker
          image: apachepulsar/pulsar:latest
          command: ["sh", "-c"]
          args:
            - >
              bin/apply-config-from-env.py conf/bookkeeper.conf &&
              bin/bookkeeper autorecovery
          envFrom:
            - configMapRef:
                name: bookie-config
          env:
            ## Configure for lower mem usage
            - name: BOOKIE_MEM
              value: '" -Xmx256m "'
            - name: BOOKIE_GC
              value: '"  "'
# Uncomment the following if bookie should be deployed on a specific instance type
#            nodeSelector:
#              beta.kubernetes.io/instance-type: m4.2xlarge

---

broker.yaml

apiVersion: v1
kind: ConfigMap
metadata:
  namespace: pulsar
  name: broker-config
data:
  # Tune for available memory. Increase the heap up to 24G to have
  # better GC behavior at high throughput
  PULSAR_MEM: '" -Dio.netty.leakDetectionLevel=disabled -Dio.netty.recycler.linkCapacity=1024 -XX:+ParallelRefProcEnabled -XX:+UnlockExperimentalVMOptions -XX:+AggressiveOpts -XX:+DoEscapeAnalysis -XX:ParallelGCThreads=32 -XX:ConcGCThreads=32 -XX:G1NewSizePercent=50 -XX:+DisableExplicitGC -XX:-ResizePLAB -XX:+ExitOnOutOfMemoryError -XX:+PerfDisableSharedMem -Xms512m -Xmx512m -XX:MaxDirectMemorySize=1g -Dpulsar.root.logger=DEBUG,FILE "'
  PULSAR_GC: '" -XX:+UseG1GC -XX:MaxGCPauseMillis=10"'
  zookeeperServers: zk-0.zookeeper,zk-1.zookeeper,zk-2.zookeeper
  configurationStoreServers: zk-0.zookeeper,zk-1.zookeeper,zk-2.zookeeper
  clusterName: eu-central-1
  managedLedgerDefaultEnsembleSize: "2"
  managedLedgerDefaultWriteQuorum: "2"
  managedLedgerDefaultAckQuorum: "2"
  deduplicationEnabled: "false"
---
##
## Broker deployment definition
##
apiVersion: apps/v1
kind: Deployment
metadata:
  namespace: pulsar
  name: broker
  labels:
    app: pulsar
    component: broker
    role: broker
spec:
  # Must match template:metdata:labels
  selector:
      matchLabels:
          app: pulsar
          component: broker
          role: broker
  replicas: 1
  template:
    metadata:
      labels:
        app: pulsar
        component: broker
        role: broker
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "8080"
    spec:
      # Enable hostNetwork only if broker pod should use the host ip
      #            hostNetwork: true
      #            dnsPolicy: ClusterFirstWithHostNet
      containers:
        - name: broker
          image: apachepulsar/pulsar:latest
          command: ["sh", "-c"]
          args:
            - >
              mkdir logs &&
              bin/apply-config-from-env.py conf/broker.conf &&
              bin/apply-config-from-env.py conf/pulsar_env.sh &&
              bin/pulsar broker
          ports:
            - containerPort: 8080
              # hostPort: 8080
            - containerPort: 6650
              # hostPort: 6650
          envFrom:
            - configMapRef:
                name: broker-config
          env:
            - name: advertisedAddress
              valueFrom:
                fieldRef:
                  fieldPath: status.podIP
# Uncomment the following if broker should be deployed on a specific instance type
#            nodeSelector:
#              beta.kubernetes.io/instance-type: m4.2xlarge
---
##
## Define the Broker headless service
## In practice, in this case, it is only useful to have a view of
## all the broker pods that are present
##
apiVersion: v1
kind: Service
metadata:
  namespace: pulsar
  name: broker
  labels:
    app: pulsar
    component: broker
    role: broker
spec:
  ports:
    - port: 8080
      name: http
    - port: 6650
      name: pulsar
  clusterIP: None
  selector:
    app: pulsar
    component: broker

---
###

apiVersion: v1
kind: Pod
metadata:
  namespace: pulsar
  name: pulsar-admin
  labels:
    app: pulsar
    role: admin
spec:
  containers:
    - name: pulsar-admin
      image: apachepulsar/pulsar:latest
      command: ["sh", "-c"]
      args:
        - >
          bin/apply-config-from-env.py conf/client.conf &&
          sleep 10000000000
      envFrom:
        - configMapRef:
            name: broker-config
      env:
        - name: webServiceUrl
          value: http://broker:8080/
        - name: brokerServiceUrl
          value: pulsar://broker:6650/

cluster-metadata.yaml

apiVersion: batch/v1
kind: Job
metadata:
  namespace: pulsar
  name: pulsar-cluster-metadata-init
  labels:
    app: pulsar
    component: broker
spec:
  template:
    spec:
      containers:
        - name: pulsar-cluster-metadata-init-container
          image: apachepulsar/pulsar:latest
          command: ["sh", "-c"]
          args:
            - >
              bin/pulsar initialize-cluster-metadata \
                --cluster eu-central-1 \
                --zookeeper zookeeper \
                --configuration-store zookeeper \
                --web-service-url http://broker:8080/ \
                --broker-service-url pulsar://broker:6650/ || true;
      restartPolicy: Never

proxy.yaml

apiVersion: v1
kind: ConfigMap
metadata:
  namespace: pulsar
  name: pulsar-proxy-config
data:
  PULSAR_MEM: '" -Xms256m -Xmx256m -XX:MaxDirectMemorySize=4g"'
  brokerServiceURL: pulsar://broker:6650
  brokerWebServiceURL: http://broker:8080
  clusterName: eu-central-1
  zookeeperServers: zk-0.zookeeper,zk-1.zookeeper,zk-2.zookeeper
  configurationStoreServers: zk-0.zookeeper,zk-1.zookeeper,zk-2.zookeeper
---
apiVersion: apps/v1
kind: Deployment
metadata:
  namespace: pulsar
  name: broker-proxy
  labels:
    app: pulsar
    component: proxy
spec:
  # Must match template:metdata:labels
  selector:
      matchLabels:
          app: pulsar
          component: proxy
  replicas: 3
  template:
    metadata:
      labels:
        app: pulsar
        component: proxy
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "8080"
    spec:
      containers:
        - name: pulsar-proxy
          image: apachepulsar/pulsar:latest
          command: ["sh", "-c"]
          args:
            - >
              bin/apply-config-from-env.py conf/proxy.conf &&
              bin/apply-config-from-env.py conf/pulsar_env.sh &&
              bin/pulsar proxy
          ports:
            - containerPort: 6650
            - containerPort: 8080
          envFrom:
            - configMapRef:
                name: pulsar-proxy-config
---
# apiVersion: v1
# kind: Service
# metadata:
#  namespace: pulsar
#   name: broker-proxy
#   labels:
#     app: pulsar
#     component: proxy
# spec:
#   type: LoadBalancer
#   ports:
#     - name: pulsar
#       port: 6650
#       protocol: TCP
#     - name: http
#       port: 8080
#       protocol: TCP
#   selector:
#     app: pulsar
#     component: proxy

zookeeper.yaml

apiVersion: v1
kind: ConfigMap
metadata:
  namespace: pulsar
  name: zookeeper-config
data:
  PULSAR_MEM: '" -Dcom.sun.management.jmxremote -Djute.maxbuffer=10485760 -Xms128m -Xmx256m  -XX:+ParallelRefProcEnabled -XX:+UnlockExperimentalVMOptions -XX:+AggressiveOpts -XX:+DoEscapeAnalysis -XX:+DisableExplicitGC -XX:+PerfDisableSharedMem  -Dzookeeper.forceSync=no "'
  PULSAR_GC: '" -XX:+UseG1GC -XX:MaxGCPauseMillis=10"'
---
## Define a disruption budget to ensure there are at least
## 2 ZK servers running all the time
apiVersion: policy/v1beta1
kind: PodDisruptionBudget
metadata:
  namespace: pulsar
  name: zk-budget
spec:
  selector:
    matchLabels:
      app: zk
  minAvailable: 2
---
## Define a StatefulSet for ZK servers
apiVersion: apps/v1
kind: StatefulSet
metadata:
  namespace: pulsar
  name: zk
  labels:
    app: pulsar
    component: zookeeper
spec:
  serviceName: zookeeper
  # Must match template:metadata:labels
  selector:
      matchLabels:
          app: pulsar
          component: zookeeper
          cluster: eu-central-1
  replicas: 3
  template:
    metadata:
      labels:
        app: pulsar
        component: zookeeper
        cluster: eu-central-1
      annotations:
        pod.alpha.kubernetes.io/initialized: "true"
        prometheus.io/scrape: "true"
        prometheus.io/port: "8000"

    spec:
      # Make sure multiple pods of ZK don't get scheduled on the
      # same node, unless there are no other available nodes
      affinity:
        podAntiAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
            - weight: 1
              podAffinityTerm:
                labelSelector:
                  matchExpressions:
                    - key: "app"
                      operator: In
                      values:
                        - pulsar
                    - key: "component"
                      operator: In
                      values:
                        - zookeeper
                topologyKey: "kubernetes.io/hostname"
      containers:
        - name: zookeeper
          image: apachepulsar/pulsar:latest
          command: ["sh", "-c"]
          args:
            - >
              bin/apply-config-from-env.py conf/zookeeper.conf &&
              bin/apply-config-from-env.py conf/pulsar_env.sh &&
              bin/generate-zookeeper-config.sh conf/zookeeper.conf &&
              bin/pulsar zookeeper
          ports:
            - containerPort: 2181
              name: client
            - containerPort: 2888
              name: server
            - containerPort: 3888
              name: leader-election
          env:
            - name: ZOOKEEPER_SERVERS
              value: zk-0,zk-1,zk-2
          envFrom:
            - configMapRef:
                name: zookeeper-config

          readinessProbe:
            exec:
              command:
                - "bin/pulsar-zookeeper-ruok.sh"
            initialDelaySeconds: 5
            timeoutSeconds: 5

          livenessProbe:
            exec:
              command:
                - "bin/pulsar-zookeeper-ruok.sh"
            initialDelaySeconds: 15
            timeoutSeconds: 5

          volumeMounts:
            - name: datadir
              mountPath: /pulsar/data
      volumes:
        - name: datadir
          emptyDir: {}
# Uncomment the following if zk should be deployed on a specific instance type
#            nodeSelector:
#              beta.kubernetes.io/instance-type: t2.large

---
##
## Define the ZooKeeper headless service
##
apiVersion: v1
kind: Service
metadata:
  namespace: pulsar
  annotations:
    service.alpha.kubernetes.io/tolerate-unready-endpoints: "true"
  name: zookeeper
  labels:
    app: pulsar
    component: zookeeper
spec:
  ports:
    - port: 2888
      name: server
    - port: 3888
      name: leader-election
    - port: 2181
      name: stats
  clusterIP: None
  selector:
    app: pulsar
    component: zookeeper