metadata.yaml

# yaml-language-server: $schema=../../component-metadata-schema.json
schemaVersion: v1
type: bindings
name: kafka
version: v1
status: stable
title: "Apache Kafka"
urls:
  - title: Reference
    url: https://docs.dapr.io/reference/components-reference/supported-bindings/kafka/
binding:
  output: true
  input: true
  operations:
    - name: create
      description: "Publish a new message in the topic."
# This auth profile has duplicate fields intentionally as we maintain backwards compatibility,
# but also move Kafka to utilize the noramlized AWS fields in the builtin auth profiles.
# TODO: rm the duplicate aws prefixed fields in Dapr 1.17.
builtinAuthenticationProfiles:
  - name: "aws"
    metadata:
      - name: authType
        type: string
        required: true
        description: |
          Authentication type.
          This must be set to "awsiam" for this authentication profile.
        example: '"awsiam"'
        allowedValues:
          - "awsiam"
      - name: awsAccessKey
        type: string
        required: false
        description: |
          This maintains backwards compatibility with existing fields.
          It will be deprecated as of Dapr 1.17. Use 'accessKey' instead.
          If both fields are set, then 'accessKey' value will be used.
          AWS access key associated with an IAM account.
        example: '"AKIAIOSFODNN7EXAMPLE"'
      - name: awsSecretKey
        type: string
        required: false
        sensitive: true
        description: |
          This maintains backwards compatibility with existing fields.
          It will be deprecated as of Dapr 1.17. Use 'secretKey' instead.
          If both fields are set, then 'secretKey' value will be used.
          The secret key associated with the access key.
        example: '"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"'
      - name: awsSessionToken
        type: string
        sensitive: true
        description: |
          This maintains backwards compatibility with existing fields.
          It will be deprecated as of Dapr 1.17. Use 'sessionToken' instead.
          If both fields are set, then 'sessionToken' value will be used.
          AWS session token to use. A session token is only required if you are using temporary security credentials.
        example: '"TOKEN"'
      - name: awsIamRoleArn
        type: string
        required: false
        description: |
          This maintains backwards compatibility with existing fields.
          It will be deprecated as of Dapr 1.17. Use 'assumeRoleArn' instead.
          If both fields are set, then 'assumeRoleArn' value will be used.
          IAM role that has access to MSK. This is another option to authenticate with MSK aside from the AWS Credentials.
        example: '"arn:aws:iam::123456789:role/mskRole"'
      - name: awsStsSessionName
        type: string
        description: |
          This maintains backwards compatibility with existing fields.
          It will be deprecated as of Dapr 1.17. Use 'sessionName' instead.
          If both fields are set, then 'sessionName' value will be used.
          Represents the session name for assuming a role.
        example: '"MyAppSession"'
        default: '"DaprDefaultSession"'
authenticationProfiles:
  - title: "OIDC Authentication"
    description: |
      Authenticate using OpenID Connect.
    metadata:
      - name: authType
        type: string
        required: true
        description: |
          Authentication type.
          This must be set to "oidc" for this authentication profile.
        example: '"oidc"'
        allowedValues:
          - "oidc"
      - name: oidcTokenEndpoint
        type: string
        required: true
        description: |
          URL of the OAuth2 identity provider access token endpoint.
        example: '"https://identity.example.com/v1/token"'
      - name: oidcClientID
        description: |
          The OAuth2 client ID that has been provisioned in the identity provider.
        example: '"my-client-id"'
        type: string
        required: true
      - name: oidcClientSecret
        type: string
        required: true
        sensitive: true
        description: |
          The OAuth2 client secret that has been provisioned in the identity provider.
        example: '"KeFg23!"'
      - name: oidcScopes
        type: string
        description: |
          Comma-delimited list of OAuth2/OIDC scopes to request with the access token.
          Although not required, this field is recommended.
        example: '"openid,kafka-prod"'
        default: '"openid"'
      - name: oidcExtensions
        description: |
          String containing a JSON-encoded dictionary of OAuth2/OIDC extensions to request with the access token.
        example: |
          {"cluster":"kafka","poolid":"kafkapool"}
        type: string
      - name: oidcClientAuthMethod
        type: string
        required: false
        default: '"client_secret"'
        description: |
          Client authentication method at the token endpoint.
          The "client_secret" method (default) uses the client_id and client_secret to authenticate.
          The "client_jwt" method uses a JWT client assertion to authenticate.
        example: '"client_jwt"'
        allowedValues:
          - "client_secret"
          - "client_jwt"
      - name: oidcClientAssertionCert
        type: string
        required: false
        description: |
          Required if "oidcClientAuthMethod" is set to "client_jwt".
          PEM-encoded X.509 certificate used to advertise the key in the x5c header when using client_jwt.
        example: |
          -----BEGIN CERTIFICATE-----\n...
      - name: oidcClientAssertionKey
        type: string
        required: false
        sensitive: true
        description: |
          Required if "oidcClientAuthMethod" is set to "client_jwt".
          PEM-encoded private key used to sign JWT client assertions for client_jwt.
        example: |
          -----BEGIN PRIVATE KEY-----\n...
      - name: oidcResource
        type: string
        required: false
        description: |
          Primarily used with "client_jwt". Optional OAuth2 resource (audience) parameter
          to include in the token request when required by the identity provider.
        example: '"api://kafka"'
      - name: oidcAudience
        type: string
        required: false
        description: |
          Only used if "oidcClientAuthMethod" is set to "client_jwt".
          Overrides the JWT client assertion audience (aud). If not set, the component uses the
          issuer derived from the token endpoint URL when available; otherwise, it falls back to the token URL.
        example: '"http://<idp-host>/realms/local"'
  - title: "SASL Authentication"
    description: |
      Authenticate using SASL.
    metadata:
      - name: authType
        type: string
        required: true
        description: |
          Authentication type.
          This must be set to "password" for this authentication profile.
        example: '"password"'
        allowedValues:
          - "password"
      - name: saslUsername
        type: string
        required: true
        description: |
          The SASL username.
        example: '"myuser"'
      - name: saslPassword
        type: string
        required: true
        sensitive: true
        description: |
          The SASL password.
        example: '"mypassword"'
      - name: saslMechanism
        type: string
        required: true
        description: |
          The SASL authentication mechanism to use.
        example: '"SHA-512"'
        default: '"PLAINTEXT"'
        allowedValues:
          - "SHA-512"
          - "SHA-256"
          - "PLAINTEXT"
  - title: "mTLS Authentication"
    description: |
      Authenticate using mTLS.
    metadata:
      - name: authType
        type: string
        required: true
        description: |
          Authentication type.
          This must be set to "mtls" for this authentication profile.
        example: '"mtls"'
        allowedValues:
          - "mtls"
      - name: caCert
        type: string
        required: true
        description: "Certificate authority certificate."
        example: "-----BEGIN CERTIFICATE-----\n<base64-encoded DER>\n-----END CERTIFICATE-----"
      - name: clientCert
        required: true
        description: "Client certificate."
        example: "-----BEGIN CERTIFICATE-----\n<base64-encoded DER>\n-----END CERTIFICATE-----"
        type: string
      - name: clientKey
        type: string
        required: true
        sensitive: true
        description: "Client key."
        example: "-----BEGIN RSA PRIVATE KEY-----\n<base64-encoded DER>\n-----END RSA PRIVATE KEY-----"
  - title: "No Authentication"
    description: |
      Do not perform authentication.
    metadata:
      - name: authType
        type: string
        required: true
        description: |
          Authentication type.
          This must be set to "none" for this authentication profile.
        example: '"none"'
        allowedValues:
          - "none"
metadata:
  - name: topics
    type: string
    description: |
      A comma-separated list of topics to subscribe to.
    example: '"mytopic1,topic2"'
    binding:
      input: true
  - name: brokers
    type: string
    required: true
    description: |
      A comma-separated list of Kafka brokers.
    example: '"localhost:9092,dapr-kafka.myapp.svc.cluster.local:9093"'
    binding:
      output: true
      input: true
  - name: publishTopic
    type: string
    required: true
    description: |
      The topic to publish to.
    example: '"mytopic"'
    binding:
      output: true
  - name: consumerGroup
    type: string
    description: |
      A kafka consumer group to listen on. Each record published
      to a topic is delivered to one consumer within each consumer
      group subscribed to the topic.
    example: '"group1"'
    binding:
      input: true
  - name: clientConnectionTopicMetadataRefreshInterval
    type: duration
    description: |
      The interval for the client connection's topic metadata to be refreshed with the broker as a Go duration.
    example: '4m'
    default: '9m'
  - name: clientConnectionKeepAliveInterval
    type: duration
    description: |
      The max amount of time for the client connection to be kept alive with the broker, as a Go duration, before closing the connection. A zero value (default) means keeping alive indefinitely.
    example: '4m'
    default: '0'
  - name: clientID
    type: string
    description: |
      A user-provided string sent with every request to
      the Kafka brokers for logging, debugging, and auditing purposes.
    example: '"my-dapr-app"'
    default: '"sarama"'
  - name: initialOffset
    type: string
    description: |
      The initial offset to use if no offset was previously committed.
    example: '"oldest"'
    default: '"newest"'
    allowedValues:
      - "newest"
      - "oldest"
    binding:
      input: true
  - name: maxMessageBytes
    type: number
    description: |
      The maximum size in bytes allowed for a single Kafka message.
    example: '2048'
    default: '1024'
  - name: consumeRetryInterval
    type: duration
    description: |
      The interval between retries when attempting to consume topics.
    example: '"200ms"'
    default: '"100ms"'
  - name: consumeRetryEnabled
    type: bool
    description: |
      Disables consumer retry by setting this to "false".
    example: '"true"'
    default: '"false"'
  - name: heartbeatInterval
    type: duration
    description: |
      The interval between heartbeats to the consumer coordinator.
    example: '"5s"'
    default: '"3s"'
  - name: sessionTimeout
    type: duration
    description: |
      The maximum time between heartbeats before the consumer is considered inactive and will timeout.
    example: '"20s"'
    default: '"10s"'
  - name: version
    type: string
    description: |
      Kafka cluster version.
      Note that this must be set to "1.0.0" if you are using Azure Event Hubs with Kafka.
    example: '"0.10.2.0"'
    default: '"2.0.0.0"'
  - name: skipVerify
    required: false
    description: |
      Skip TLS verification.
      This is potentially insecure and not recommended for use in production.
    example: "true"
    default: "false"
    type: bool
  - name: disableTls
    required: false
    description: |
      Disable TLS for transport security.
      This is potentially insecure and not recommended for use in production.
    example: "true"
    default: "false"
    type: bool
  - name: schemaRegistryURL
    type: string
    description: |
      The Schema Registry URL.
    example: '"http://localhost:8081"'
  - name: schemaRegistryAPIKey
    type: string
    description: |
      The Schema Registry credentials API Key.
    example: '"XYAXXAZ"'
  - name: schemaRegistryAPISecret
    type: string
    description: |
      The Schema Registry credentials API Secret.
    example: '"ABCDEFGMEADFF"'
  - name: schemaCachingEnabled
    type: bool
    description: |
      Enables caching for schemas.
    example: '"true"'
    default: '"true"'
  - name: schemaLatestVersionCacheTTL
    type: duration
    description: |
      The TTL for schema caching when publishing a message with latest schema available.
    example: '"5m"'
    default: '"5m"'
  - name: escapeHeaders
    type: bool
    required: false
    description: |
      Enables URL escaping of the message header values.
      It allows sending headers with special characters that are usually not allowed in HTTP headers.
    example: "true"
    default: "false"
  - name: useAvroJSON
    type: bool
    required: false
    description: |
      Enables Avro JSON schema for serialization. Only applicable when the subscription uses valueSchemaType=Avro
    example: "true"
    default: "false"
  - name: compression
    type: string
    required: false
    description: |
      Enables message compression.
      There are five types of compression available: none, gzip, snappy, lz4, and zstd.
      The default is none.
    example: '"gzip"'
    default: "none"
  - name: consumerGroupRebalanceStrategy
    type: string
    required: false
    description: |
      The strategy to use for consumer group rebalancing.
    example: '"sticky"'
    default: '"range"'
    allowedValues:
      - "range"
      - "sticky"
      - "roundrobin"
  - name: excludeHeaderMetaRegex
    type: string
    required: false
    description: |
      A regular expression to exclude keys from being converted to/from headers from/to metadata to avoid unwanted downstream side effects.
    example: '"^rawPayload|valueSchemaType$"'
    default: '""'