kafka-topic

- uri: template://default-provisioners/kafka-topic
  type: kafka-topic
  description: Provisions a dedicated Kafka topic on a shared Kafka broker.
  init: |
    brokerPort: 9092
    ctrlPort: 9093    
  state: |
    topic: {{ dig "topic" (print "topic-" (randAlphaNum 6)) .State | quote }}    
  shared: |
    shared_kafka_instance_name: {{ dig "shared_kafka_instance_name" (print "kafka-" (randAlphaNum 6)) .Shared | quote }}    
  services: |
    {{ .Shared.shared_kafka_instance_name }}:
      image: bitnami/kafka:latest
      restart: always
      environment:
        KAFKA_CFG_NODE_ID: "0"
        KAFKA_CFG_PROCESS_ROLES: controller,broker
        KAFKA_CFG_LISTENERS: "PLAINTEXT://:{{ .Init.brokerPort }},CONTROLLER://:{{ .Init.ctrlPort }}"
        KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"
        KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: "0@{{ .Shared.shared_kafka_instance_name }}:{{ .Init.ctrlPort }}"
        KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
        KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: "false"
      healthcheck:
        test: ["CMD", "kafka-topics.sh", "--list", "--bootstrap-server=localhost:{{ .Init.brokerPort }}"]
        interval: 2s
        timeout: 2s
        retries: 10
      {{ $publishPort := (dig "annotations" "compose.score.dev/publish-port" "0" .Metadata | atoi) }}
      {{ if ne $publishPort 0 }}
      ports:
      - target: {{ .Init.brokerPort }}
        published: {{ $publishPort }}
      {{ end }}
      volumes:
      - type: volume
        source: {{ .Shared.shared_kafka_instance_name }}-data
        target: /bitnami/kafka
    {{ .State.topic }}-init:
      image: bitnami/kafka:latest
      entrypoint: ["/bin/sh"]
      command: ["-c", "kafka-topics.sh --topic={{.State.topic}} --bootstrap-server=localhost:{{ .Init.brokerPort }} --describe || kafka-topics.sh --topic={{.State.topic}} --bootstrap-server=localhost:{{ .Init.brokerPort }} --create --partitions=3"]
      network_mode: "service:{{ .Shared.shared_kafka_instance_name }}"
      labels:
        dev.score.compose.labels.is-init-container: "true"
      depends_on:
        {{ .Shared.shared_kafka_instance_name }}:
          condition: service_healthy
          restart: true    
  volumes: |
    {{ .Shared.shared_kafka_instance_name }}-data:
      driver: local    
  outputs: |
    host: {{ .Shared.shared_kafka_instance_name }}
    port: "{{ .Init.brokerPort }}"
    name: {{ .State.topic }}
    num_partitions: 3    
  expected_outputs:
    - host
    - port
    - name
    - num_partitions