Skip to main content
Quick navigation

Kafka resources

Kafka resources​

Topic​

Creates a Topic in Kafka.

API Keys: Admin API Key Application API Key
Managed with: CLI API Terraform Console UI
Labels support: Full Label Support

---
apiVersion: kafka/v2
kind: Topic
metadata:
  cluster: shadow-it
  name: click.event-stream.avro
  labels:
    domain: clickstream
    appcode: clk
  description: | 
    # Event Stream from Click Application
    This is a multiline markdown description that will appear in the Topic Catalog
  descriptionIsEditable: false
  catalogVisibility: PUBLIC
spec:
  replicationFactor: 3
  partitions: 3
  configs:
    min.insync.replicas: '2'
    cleanup.policy: delete
    retention.ms: '60000'

Topic checks:

  • metadata.cluster is a valid Kafka Cluster
  • metadata.name must belong to the Application Instances
  • spec.replicationFactor and spec.partitions are immutable and cannot be modified once the Topic is created
  • spec.configs must be valid Kafka Topic configs
  • All properties are validated against TopicPolicies attached to the Application Instance

Conduktor annotations:

  • metadata.description is optional. The description field in markdown that will be displayed in the Topic Catalog view
    • Previously conduktor.io/description.editable in 1.28 and below
  • metadata.descriptionIsEditable is optional (defaults "true"). Defines whether the description can be updated in the UI
    • Previously conduktor.io/description.editable in 1.28 and below
  • metadata.catalogVisibility is optional. Can be PUBLIC or PRIVATE.
    • When the topic is linked to a Self-Service Application, defines whether the topic is visible (PUBLIC) in the Topic Catalog or not (PRIVATE).
    • If empty, the Topic Catalog Visibility is inherited from the ApplicationInstance field spec.defaultCatalogVisibility.

Side effects

Subject​

Creates a Subject in the Schema Registry.

API Keys: Admin API Key Application API Key
Managed with: CLI API Console UI
Labels support: Partial Label Support (No UI yet)

Local file

---
apiVersion: kafka/v2
kind: Subject
metadata:
  cluster: shadow-it
  name: myPrefix.topic-value
spec:
  schemaFile: schemas/topic.avsc # relative to Conduktor CLI execution context
  format: AVRO
  compatibility: FORWARD_TRANSITIVE

Inline

---
apiVersion: kafka/v2
kind: Subject
metadata:
  cluster: shadow-it
  name: myPrefix.topic-value
spec:
  schema: |
    {
      "type": "long"
    }
  format: AVRO

Schema reference

---
apiVersion: kafka/v2
kind: Subject
metadata:
  cluster: shadow-it
  name: myPrefix.topic-value
spec:
  schema: |
    {
      "type": "record",
      "namespace": "com.schema.avro",
      "name": "Client",
      "fields": [
        {
          "name": "name",
          "type": "string"
        },
        {
          "name": "address",
          "type": "com.schema.avro.Address"
        }
      ]
    }
  format: AVRO
  references:
    - name: com.schema.avro.Address
      subject: commons.address-value
      version: 1

Subject checks:

  • metadata.cluster is a valid Kafka Cluster
  • metadata.name must belong to the Application Instance
  • One of spec.schema or spec.schemaFile must be present
    • schema requires an inline schema
    • schemaFile requires a path to a file that contains the schema relative to the CLI execution path
  • spec.format is mandatory. Defines the schema format: AVRO, PROTOBUF, JSON
  • spec.compatibility is optional. Defines the subject compatibility mode: BACKWARD, BACKWARD_TRANSITIVE, FORWARD, FORWARD_TRANSITIVE, FULL, FULL_TRANSITIVE, NONE
    • Unset the field if you want the compatibility mode to be the one defined at the Schema Registry global level
  • spec.references is optional. It specifies the names of referenced schemas

Side effects

Connector​

Creates a connector on a Kafka Connect Cluster.

API Keys: Admin API Key Application API Key
Managed with: CLI API Console UI
Labels support: Partial Label Support (No UI yet)

---
apiVersion: kafka/v2
kind: Connector
metadata:
  name: click.my-connector
  cluster: 'prod-cluster'
  connectCluster: kafka-connect-cluster
  labels:
    domain: clickstream
  autoRestart:
    enabled: true
    frequencySeconds: 600
spec:
  config:
    connector.class: io.connect.jdbc.JdbcSourceConnector
    tasks.max: '1'
    topic: click.pageviews
    connection.url: "jdbc:mysql://127.0.0.1:3306/sample?verifyServerCertificate=false&useSSL=true&requireSSL=true"
    consumer.override.sasl.jaas.config: o.a.k.s.s.ScramLoginModule required username="<user>" password="<password>";

Connector checks:

  • metadata.connectCluster is a valid Kafka Connect Cluster
  • metadata.name must belong to the Application Instance

Conduktor annotations:

  • metadata.autoRestart.enabled is optional (default "false"). Defines whether the Console Automatic Restart feature is enabled for this Connector
    • Previously conduktor.io/auto-restart-enabled in 1.28 and below
  • metadata.autoRestart.frequencySeconds is optional (default 600, meaning 10 minutes). Defines the delay between consecutive restart attempts
    • Previously conduktor.io/auto-restart-frequency in 1.28 and below

Service account​

Manages the ACLs (Access Control Lists) of a service account in Kafka.

info

This doesn't create the service account, only assigns ACLs.

API Keys: Admin API Key
Managed with: CLI API Console UI
Labels support: Full Label Support

Example for Kafka service accounts:

---
apiVersion: v1
kind: ServiceAccount
metadata:
  cluster: shadow-it
  name: clickstream-sa
  labels:
    domain: clickstream
    appcode: clk
spec:
  authorization:
    type: KAFKA_ACL
    acls:
      # List all the topics
      - type: TOPIC
        name: '*'
        patternType: LITERAL
        operations:
          - Describe
      # Read & Write on the click.event-stream.avro topic
      - type: TOPIC
        name: 'click.event-stream.avro'
        patternType: LITERAL
        operations:
          - Write
          - Read
      # Read on all the topics prefixed by public_
      - type: TOPIC
        name: 'public_'
        patternType: PREFIXED
        operations:
          - Read
      # Read on the consumer groups prefixed by click.event-stream.
      - type: CONSUMER_GROUP
        name: 'click.event-stream.'
        patternType: PREFIXED
        operations:
          - Read

Example for Aiven service accounts:

---
apiVersion: v1
kind: ServiceAccount
metadata:
  cluster: aiven
  name: clickstream-sa
  labels:
    domain: clickstream
    appcode: clk
spec:
  authorization:
    type: AIVEN_ACL
    acls:
      # Read & Write on the click.event-stream.avro topic
      - resourceType: TOPIC
        name: 'click.event-stream.avro'
        permission: readwrite
      # Read on all the topics prefixed by public_
      - type: TOPIC
        name: 'public*'
        permission: read
      # Write on the click.event-stream.avro schema
      - type: SCHEMA
        name: 'Subject:click.event-stream.avro'
        permission: schema_registry_write

Service account checks:

  • metadata.cluster is a valid Kafka Cluster.
  • metadata.name is a valid, pre-existing service account.
  • spec.authorization.type must be 'KAFKA_ACL' or 'AIVEN_ACL'. 'AIVEN_ACL' is only supported for Aiven Kafka clusters. 'KAFKA_ACL' is not supported for Aiven Kafka clusters.

When spec.authorization.type equals KAFKA_ACL:

  • spec.acls[].type must be a valid resource type on Kafka (Kafka ACL Operations and Resources)
  • spec.acls[].operations must contain only operations that are valid for the resource type.
  • spec.acls[].host is optional, and will default to '*'.
  • spec.acls[].permission is optional, and will default to 'Allow'.

When spec.authorization.type equals AIVEN_ACL:

  • spec.acls[].resourceType must be a valid resource type on Aiven Kafka: TOPIC (Aiven ACL for topics) or SCHEMA (Aiven ACL for schema)
  • spec.acls[].name must be a valid resource name on Aiven Kafka. For schemas, it must match ^(Config:|Subject:[A-Za-z0-9/_.*?-]+).
  • spec.acls[].permission must contain only operations that are valid for the resource type.

Side effects

  • Kafka:
    • Service account ACLs are created/updated.
    • In dry-run mode, service account ACLs are validated against the aforementioned criteria, ensuring the ACL definitions are legal.