Pub/sub brokers component specs

• 1: Apache Kafka
• 2: AWS SNS/SQS
• 3: Azure Event Hubs
• 4: Azure Service Bus Queues
• 5: Azure Service Bus Topics
• 6: GCP
• 7: In-memory
• 8: JetStream
• 9: KubeMQ
• 10: MQTT
• 11: MQTT3
• 12: Pulsar
• 13: RabbitMQ
• 14: Redis Streams
• 15: RocketMQ
• 16: Solace-AMQP

<tr>
    <td><a href="/reference/components-reference/supported-pubsub/setup-apache-kafka/">Apache Kafka</a>
    </td>
    <td>Stable</td>
    <td>v1</td>
    <td>1.5</td>
</tr>

<tr>
    <td><a href="/reference/components-reference/supported-pubsub/setup-inmemory/">In-memory</a>
    </td>
    <td>Stable</td>
    <td>v1</td>
    <td>1.7</td>
</tr>

<tr>
    <td><a href="/reference/components-reference/supported-pubsub/setup-jetstream/">JetStream</a>
    </td>
    <td>Beta</td>
    <td>v1</td>
    <td>1.10</td>
</tr>

<tr>
    <td><a href="/reference/components-reference/supported-pubsub/setup-kubemq/">KubeMQ</a>
    </td>
    <td>Beta</td>
    <td>v1</td>
    <td>1.10</td>
</tr>

<tr>
    <td><a href="/reference/components-reference/supported-pubsub/setup-mqtt3/">MQTT3</a>
    </td>
    <td>Stable</td>
    <td>v1</td>
    <td>1.7</td>
</tr>

<tr>
    <td><a href="/reference/components-reference/supported-pubsub/setup-pulsar/">Pulsar</a>
    </td>
    <td>Stable</td>
    <td>v1</td>
    <td>1.10</td>
</tr>

<tr>
    <td><a href="/reference/components-reference/supported-pubsub/setup-rabbitmq/">RabbitMQ</a>
    </td>
    <td>Stable</td>
    <td>v1</td>
    <td>1.7</td>
</tr>

<tr>
    <td><a href="/reference/components-reference/supported-pubsub/setup-redis-pubsub/">Redis Streams</a>
    </td>
    <td>Stable</td>
    <td>v1</td>
    <td>1.0</td>
</tr>

<tr>
    <td><a href="/reference/components-reference/supported-pubsub/setup-rocketmq/">RocketMQ</a>
    </td>
    <td>Alpha</td>
    <td>v1</td>
    <td>1.8</td>
</tr>

<tr>
    <td><a href="/reference/components-reference/supported-pubsub/setup-solace-amqp/">Solace-AMQP</a>
    </td>
    <td>Beta</td>
    <td>v1</td>
    <td>1.10</td>
</tr>

<tr>
    <td><a href="/reference/components-reference/supported-pubsub/setup-aws-snssqs/">AWS SNS/SQS</a>
    </td>
    <td>Stable</td>
    <td>v1</td>
    <td>1.10</td>
</tr>

<tr>
    <td><a href="/reference/components-reference/supported-pubsub/setup-gcp-pubsub/">GCP Pub/Sub</a>
    </td>
    <td>Stable</td>
    <td>v1</td>
    <td>1.11</td>
</tr>

<tr>
    <td><a href="/reference/components-reference/supported-pubsub/setup-azure-eventhubs/">Azure Event Hubs</a>
    </td>
    <td>Stable</td>
    <td>v1</td>
    <td>1.8</td>
</tr>

<tr>
    <td><a href="/reference/components-reference/supported-pubsub/setup-azure-servicebus-queues/">Azure Service Bus Queues</a>
    </td>
    <td>Beta</td>
    <td>v1</td>
    <td>1.10</td>
</tr>

<tr>
    <td><a href="/reference/components-reference/supported-pubsub/setup-azure-servicebus-topics/">Azure Service Bus Topics</a>
    </td>
    <td>Stable</td>
    <td>v1</td>
    <td>1.0</td>
</tr>

1 - Apache Kafka

Component format

To set up Apache Kafka pub/sub, create a component of type pubsub.kafka. See the pub/sub broker component file to learn how ConsumerID is automatically generated. Read the How-to: Publish and Subscribe guide on how to create and apply a pub/sub configuration.

All component metadata field values can carry templated metadata values, which are resolved on Dapr sidecar startup. For example, you can choose to use {namespace} as the consumerGroup to enable using the same appId in different namespaces using the same topics as described in this article.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: kafka-pubsub
spec:
  type: pubsub.kafka
  version: v1
  metadata:
  - name: brokers # Required. Kafka broker connection setting
    value: "dapr-kafka.myapp.svc.cluster.local:9092"
  - name: consumerGroup # Optional. Used for input bindings.
    value: "{namespace}"
  - name: consumerID # Optional. If not supplied, runtime will create one.
    value: "channel1"
  - name: clientID # Optional. Used as client tracing ID by Kafka brokers.
    value: "my-dapr-app-id"
  - name: authType # Required.
    value: "password"
  - name: saslUsername # Required if authType is `password`.
    value: "adminuser"
  - name: saslPassword # Required if authType is `password`.
    secretKeyRef:
      name: kafka-secrets
      key: saslPasswordSecret
  - name: saslMechanism
    value: "SHA-512"
  - name: maxMessageBytes # Optional.
    value: 1024
  - name: consumeRetryInterval # Optional.
    value: 200ms
  - name: heartbeatInterval # Optional.
    value: 5s
  - name: sessionTimeout # Optional.
    value: 15s
  - name: version # Optional.
    value: 2.0.0
  - name: disableTls # Optional. Disable TLS. This is not safe for production!! You should read the `Mutual TLS` section for how to use TLS.
    value: "true"
  - name: consumerFetchMin # Optional. Advanced setting. The minimum number of message bytes to fetch in a request - the broker will wait until at least this many are available.
    value: 1
  - name: consumerFetchDefault # Optional. Advanced setting. The default number of message bytes to fetch from the broker in each request.
    value: 2097152
  - name: channelBufferSize # Optional. Advanced setting. The number of events to buffer in internal and external channels.
    value: 512
  - name: consumerGroupRebalanceStrategy # Optional. Advanced setting. The strategy to use for consumer group rebalancing.
    value: sticky
  - name: schemaRegistryURL # Optional. When using Schema Registry Avro serialization/deserialization. The Schema Registry URL.
    value: http://localhost:8081
  - name: schemaRegistryAPIKey # Optional. When using Schema Registry Avro serialization/deserialization. The Schema Registry API Key.
    value: XYAXXAZ
  - name: schemaRegistryAPISecret # Optional. When using Schema Registry Avro serialization/deserialization. The Schema Registry credentials API Secret.
    value: "ABCDEFGMEADFF"
  - name: schemaCachingEnabled # Optional. When using Schema Registry Avro serialization/deserialization. Enables caching for schemas.
    value: true
  - name: schemaLatestVersionCacheTTL # Optional. When using Schema Registry Avro serialization/deserialization. The TTL for schema caching when publishing a message with latest schema available.
    value: 5m
  - name: useAvroJson # Optional. Enables Avro JSON schema for serialization as opposed to Standard JSON default. Only applicable when the subscription uses valueSchemaType=Avro
    value: "true"
  - name: escapeHeaders # Optional.
    value: false
  

For details on using secretKeyRef, see the guide on how to reference secrets in components.

Spec metadata fields

The secretKeyRef above is referencing a kubernetes secrets store to access the tls information. Visit here to learn more about how to configure a secret store component.

Note

The metadata version must be set to 1.0.0 when using Azure EventHubs with Kafka.

Authentication

Kafka supports a variety of authentication schemes and Dapr supports several: SASL password, mTLS, OIDC/OAuth2. With the added authentication methods, the authRequired field has been deprecated from the v1.6 release and instead the authType field should be used. If authRequired is set to true, Dapr will attempt to configure authType correctly based on the value of saslPassword. The valid values for authType are:

• none
• password
• certificate
• mtls
• oidc
• awsiam

None

Setting authType to none will disable any authentication. This is NOT recommended in production.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: kafka-pubsub-noauth
spec:
  type: pubsub.kafka
  version: v1
  metadata:
  - name: brokers # Required. Kafka broker connection setting
    value: "dapr-kafka.myapp.svc.cluster.local:9092"
  - name: consumerGroup # Optional. Used for input bindings.
    value: "group1"
  - name: clientID # Optional. Used as client tracing ID by Kafka brokers.
    value: "my-dapr-app-id"
  - name: authType # Required.
    value: "none"
  - name: maxMessageBytes # Optional.
    value: 1024
  - name: consumeRetryInterval # Optional.
    value: 200ms
  - name: heartbeatInterval # Optional.
    value: 5s
  - name: sessionTimeout # Optional.
    value: 15s
  - name: version # Optional.
    value: 0.10.2.0
  - name: disableTls
    value: "true"

SASL Password

Setting authType to password enables SASL authentication. This requires setting the saslUsername and saslPassword fields.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: kafka-pubsub-sasl
spec:
  type: pubsub.kafka
  version: v1
  metadata:
  - name: brokers # Required. Kafka broker connection setting
    value: "dapr-kafka.myapp.svc.cluster.local:9092"
  - name: consumerGroup # Optional. Used for input bindings.
    value: "group1"
  - name: clientID # Optional. Used as client tracing ID by Kafka brokers.
    value: "my-dapr-app-id"
  - name: authType # Required.
    value: "password"
  - name: saslUsername # Required if authType is `password`.
    value: "adminuser"
  - name: saslPassword # Required if authType is `password`.
    secretKeyRef:
      name: kafka-secrets
      key: saslPasswordSecret
  - name: saslMechanism
    value: "SHA-512"
  - name: maxMessageBytes # Optional.
    value: 1024
  - name: consumeRetryInterval # Optional.
    value: 200ms
  - name: heartbeatInterval # Optional.
    value: 5s
  - name: sessionTimeout # Optional.
    value: 15s
  - name: version # Optional.
    value: 0.10.2.0
  - name: caCert
    secretKeyRef:
      name: kafka-tls
      key: caCert

Mutual TLS

Setting authType to mtls uses a x509 client certificate (the clientCert field) and key (the clientKey field) to authenticate. Note that mTLS as an authentication mechanism is distinct from using TLS to secure the transport layer via encryption. mTLS requires TLS transport (meaning disableTls must be false), but securing the transport layer does not require using mTLS. See Communication using TLS for configuring underlying TLS transport.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: kafka-pubsub-mtls
spec:
  type: pubsub.kafka
  version: v1
  metadata:
  - name: brokers # Required. Kafka broker connection setting
    value: "dapr-kafka.myapp.svc.cluster.local:9092"
  - name: consumerGroup # Optional. Used for input bindings.
    value: "group1"
  - name: clientID # Optional. Used as client tracing ID by Kafka brokers.
    value: "my-dapr-app-id"
  - name: authType # Required.
    value: "mtls"
  - name: caCert
    secretKeyRef:
      name: kafka-tls
      key: caCert
  - name: clientCert
    secretKeyRef:
      name: kafka-tls
      key: clientCert
  - name: clientKey
    secretKeyRef:
      name: kafka-tls
      key: clientKey
  - name: maxMessageBytes # Optional.
    value: 1024
  - name: consumeRetryInterval # Optional.
    value: 200ms
  - name: heartbeatInterval # Optional.
    value: 5s
  - name: sessionTimeout # Optional.
    value: 15s
  - name: version # Optional.
    value: 0.10.2.0

OAuth2 or OpenID Connect

Setting authType to oidc enables SASL authentication via the OAUTHBEARER mechanism. This supports specifying a bearer token from an external OAuth2 or OIDC identity provider. Currently, only the client_credentials grant is supported.

Configure oidcTokenEndpoint to the full URL for the identity provider access token endpoint.

Set oidcClientID and oidcClientSecret to the client credentials provisioned in the identity provider.

If caCert is specified in the component configuration, the certificate is appended to the system CA trust for verifying the identity provider certificate. Similarly, if skipVerify is specified in the component configuration, verification will also be skipped when accessing the identity provider.

By default, the only scope requested for the token is openid; it is highly recommended that additional scopes be specified via oidcScopes in a comma-separated list and validated by the Kafka broker. If additional scopes are not used to narrow the validity of the access token, a compromised Kafka broker could replay the token to access other services as the Dapr clientID.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: kafka-pubsub
spec:
  type: pubsub.kafka
  version: v1
  metadata:
  - name: brokers # Required. Kafka broker connection setting
    value: "dapr-kafka.myapp.svc.cluster.local:9092"
  - name: consumerGroup # Optional. Used for input bindings.
    value: "group1"
  - name: clientID # Optional. Used as client tracing ID by Kafka brokers.
    value: "my-dapr-app-id"
  - name: authType # Required.
    value: "oidc"
  - name: oidcTokenEndpoint # Required if authType is `oidc`.
    value: "https://identity.example.com/v1/token"
  - name: oidcClientID      # Required if authType is `oidc`.
    value: "dapr-myapp"
  - name: oidcClientSecret  # Required if authType is `oidc`.
    secretKeyRef:
      name: kafka-secrets
      key: oidcClientSecret
  - name: oidcScopes        # Recommended if authType is `oidc`.
    value: "openid,kafka-dev"
  - name: caCert            # Also applied to verifying OIDC provider certificate
    secretKeyRef:
      name: kafka-tls
      key: caCert
  - name: maxMessageBytes # Optional.
    value: 1024
  - name: consumeRetryInterval # Optional.
    value: 200ms
  - name: heartbeatInterval # Optional.
    value: 5s
  - name: sessionTimeout # Optional.
    value: 15s
  - name: version # Optional.
    value: 0.10.2.0

AWS IAM

Authenticating with AWS IAM is supported with MSK. Setting authType to awsiam uses AWS SDK to generate auth tokens to authenticate.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: kafka-pubsub-awsiam
spec:
  type: pubsub.kafka
  version: v1
  metadata:
  - name: brokers # Required. Kafka broker connection setting
    value: "dapr-kafka.myapp.svc.cluster.local:9092"
  - name: consumerGroup # Optional. Used for input bindings.
    value: "group1"
  - name: clientID # Optional. Used as client tracing ID by Kafka brokers.
    value: "my-dapr-app-id"
  - name: authType # Required.
    value: "awsiam"
  - name: region # Required.
    value: "us-west-1"
  - name: accessKey # Optional.
    value: <AWS_ACCESS_KEY>
  - name: secretKey # Optional.
    value: <AWS_SECRET_KEY>
  - name: sessionToken # Optional.
    value: <AWS_SESSION_KEY>
  - name: assumeRoleArn # Optional.
    value: "arn:aws:iam::123456789:role/mskRole"
  - name: sessionName # Optional.
    value: "DaprDefaultSession"

Communication using TLS

By default TLS is enabled to secure the transport layer to Kafka. To disable TLS, set disableTls to true. When TLS is enabled, you can control server certificate verification using skipVerify to disable verification (NOT recommended in production environments) and caCert to specify a trusted TLS certificate authority (CA). If no caCert is specified, the system CA trust will be used. To also configure mTLS authentication, see the section under Authentication. Below is an example of a Kafka pubsub component configured to use transport layer TLS:

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: kafka-pubsub
spec:
  type: pubsub.kafka
  version: v1
  metadata:
  - name: brokers # Required. Kafka broker connection setting
    value: "dapr-kafka.myapp.svc.cluster.local:9092"
  - name: consumerGroup # Optional. Used for input bindings.
    value: "group1"
  - name: clientID # Optional. Used as client tracing ID by Kafka brokers.
    value: "my-dapr-app-id"
  - name: authType # Required.
    value: "certificate"
  - name: consumeRetryInterval # Optional.
    value: 200ms
  - name: heartbeatInterval # Optional.
    value: 5s
  - name: sessionTimeout # Optional.
    value: 15s
  - name: version # Optional.
    value: 0.10.2.0
  - name: maxMessageBytes # Optional.
    value: 1024
  - name: caCert # Certificate authority certificate.
    secretKeyRef:
      name: kafka-tls
      key: caCert
auth:
  secretStore: <SECRET_STORE_NAME>

Consuming from multiple topics

When consuming from multiple topics using a single pub/sub component, there is no guarantee about how the consumers in your consumer group are balanced across the topic partitions.

For instance, let’s say you are subscribing to two topics with 10 partitions per topic and you have 20 replicas of your service consuming from the two topics. There is no guarantee that 10 will be assigned to the first topic and 10 to the second topic. Instead, the partitions could be divided unequally, with more than 10 assigned to the first topic and the rest assigned to the second topic.

This can result in idle consumers listening to the first topic and over-extended consumers on the second topic, or vice versa. This same behavior can be observed when using auto-scalers such as HPA or KEDA.

If you run into this particular issue, it is recommended that you configure a single pub/sub component per topic with uniquely defined consumer groups per component. This guarantees that all replicas of your service are fully allocated to the unique consumer group, where each consumer group targets one specific topic.

For example, you may define two Dapr components with the following configuration:

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: kafka-pubsub-topic-one
spec:
  type: pubsub.kafka
  version: v1
  metadata:
  - name: consumerGroup
    value: "{appID}-topic-one"

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: kafka-pubsub-topic-two
spec:
  type: pubsub.kafka
  version: v1
  metadata:
  - name: consumerGroup
    value: "{appID}-topic-two"

Sending and receiving multiple messages

Apache Kafka component supports sending and receiving multiple messages in a single operation using the bulk Pub/sub API.

Configuring bulk subscribe

When subscribing to a topic, you can configure bulkSubscribe options. Refer to Subscribing messages in bulk for more details. Learn more about the bulk subscribe API.

Apache Kafka supports the following bulk metadata options:

Per-call metadata fields

Partition Key

When invoking the Kafka pub/sub, its possible to provide an optional partition key by using the metadata query param in the request url.

The param name can either be partitionKey or __key

Example:

curl -X POST http://localhost:3500/v1.0/publish/myKafka/myTopic?metadata.partitionKey=key1 \
  -H "Content-Type: application/json" \
  -d '{
        "data": {
          "message": "Hi"
        }
      }'

Message headers

All other metadata key/value pairs (that are not partitionKey or __key) are set as headers in the Kafka message. Here is an example setting a correlationId for the message.

curl -X POST http://localhost:3500/v1.0/publish/myKafka/myTopic?metadata.correlationId=myCorrelationID&metadata.partitionKey=key1 \
  -H "Content-Type: application/json" \
  -d '{
        "data": {
          "message": "Hi"
        }
      }'

Kafka Pubsub special message headers received on consumer side

When consuming messages, special message metadata are being automatically passed as headers. These are:

• __key: the message key if available
• __topic: the topic for the message
• __partition: the partition number for the message
• __offset: the offset of the message in the partition
• __timestamp: the timestamp for the message

You can access them within the consumer endpoint as follows:

• Python (FastAPI)

from fastapi import APIRouter, Body, Response, status
import json
import sys

app = FastAPI()

router = APIRouter()


@router.get('/dapr/subscribe')
def subscribe():
    subscriptions = [{'pubsubname': 'pubsub',
                      'topic': 'my-topic',
                      'route': 'my_topic_subscriber',
                      }]
    return subscriptions

@router.post('/my_topic_subscriber')
def my_topic_subscriber(
      key: Annotated[str, Header(alias="__key")],
      offset: Annotated[int, Header(alias="__offset")],
      event_data=Body()):
    print(f"key={key} - offset={offset} - data={event_data}", flush=True)
      return Response(status_code=status.HTTP_200_OK)

app.include_router(router)

Receiving message headers with special characters

The consumer application may be required to receive message headers that include special characters, which may cause HTTP protocol validation errors. HTTP header values must follow specifications, making some characters not allowed. Learn more about the protocols. In this case, you can enable escapeHeaders configuration setting, which uses URL escaping to encode header values on the consumer side.

Set escapeHeaders to true to URL escape.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: kafka-pubsub-escape-headers
spec:
  type: pubsub.kafka
  version: v1
  metadata:
  - name: brokers # Required. Kafka broker connection setting
    value: "dapr-kafka.myapp.svc.cluster.local:9092"
  - name: consumerGroup # Optional. Used for input bindings.
    value: "group1"
  - name: clientID # Optional. Used as client tracing ID by Kafka brokers.
    value: "my-dapr-app-id"
  - name: authType # Required.
    value: "none"
  - name: escapeHeaders
    value: "true"

Avro Schema Registry serialization/deserialization

You can configure pub/sub to publish or consume data encoded using Avro binary serialization, leveraging an Apache Schema Registry (for example, Confluent Schema Registry, Apicurio).

Configuration

When configuring the Kafka pub/sub component metadata, you must define:

• The schema registry URL
• The API key/secret, if applicable

Schema subjects are automatically derived from topic names, using the standard naming convention. For example, for a topic named my-topic, the schema subject will be my-topic-value. When interacting with the message payload within the service, it is in JSON format. The payload is transparently serialized/deserialized within the Dapr component. Date/Datetime fields must be passed as their Epoch Unix timestamp equivalent (rather than typical Iso8601). For example:

• 2024-01-10T04:36:05.986Z should be passed as 1704861365986 (the number of milliseconds since Jan 1st, 1970)
• 2024-01-10 should be passed as 19732 (the number of days since Jan 1st, 1970)

Publishing Avro messages

In order to indicate to the Kafka pub/sub component that the message should be using Avro serialization, the valueSchemaType metadata must be set to Avro.

• curl
• Python SDK

curl -X "POST" http://localhost:3500/v1.0/publish/pubsub/my-topic?metadata.rawPayload=true&metadata.valueSchemaType=Avro -H "Content-Type: application/json" -d '{"order_number": "345", "created_date": 1704861365986}'

from dapr.clients import DaprClient

with DaprClient() as d:
    req_data = {
        'order_number': '345',
        'created_date': 1704861365986
    }
    # Create a typed message with content type and body
    resp = d.publish_event(
        pubsub_name='pubsub',
        topic_name='my-topic',
        data=json.dumps(req_data),
        publish_metadata={'rawPayload': 'true', 'valueSchemaType': 'Avro'}
    )
    # Print the request
    print(req_data, flush=True)

Subscribing to Avro topics

In order to indicate to the Kafka pub/sub component that the message should be deserialized using Avro, the valueSchemaType metadata must be set to Avro in the subscription metadata.

• Python (FastAPI)

from fastapi import APIRouter, Body, Response, status
import json
import sys

app = FastAPI()

router = APIRouter()


@router.get('/dapr/subscribe')
def subscribe():
    subscriptions = [{'pubsubname': 'pubsub',
                      'topic': 'my-topic',
                      'route': 'my_topic_subscriber',
                      'metadata': {
                          'valueSchemaType': 'Avro',
                      } }]
    return subscriptions

@router.post('/my_topic_subscriber')
def my_topic_subscriber(event_data=Body()):
    print(event_data, flush=True)
      return Response(status_code=status.HTTP_200_OK)

app.include_router(router)

Overriding default consumer group rebalancing

In Kafka, rebalancing strategies determine how partitions are assigned to consumers within a consumer group. The default strategy is “range”, but “roundrobin” and “sticky” are also available.

• Range: Partitions are assigned to consumers based on their lexicographical order. If you have three partitions (0, 1, 2) and two consumers (A, B), consumer A might get partitions 0 and 1, while consumer B gets partition 2.
• RoundRobin: Partitions are assigned to consumers in a round-robin fashion. With the same example above, consumer A might get partitions 0 and 2, while consumer B gets partition 1.
• Sticky: This strategy aims to preserve previous assignments as much as possible while still maintaining a balanced distribution. If a consumer leaves or joins the group, only the affected partitions are reassigned, minimizing disruption.

Choosing a Strategy:

• Range: Simple to understand and implement, but can lead to uneven distribution if partition sizes vary significantly.
• RoundRobin: Provides a good balance in many cases, but might not be optimal if message keys are unevenly distributed.
• Sticky: Generally preferred for its ability to minimize disruption during rebalances, especially when dealing with a large number of partitions or frequent consumer group changes.

Create a Kafka instance

• Self-Hosted
• Kubernetes

Related links

• Basic schema for a Dapr component
• Read this guide for instructions on configuring pub/sub components
• Pub/Sub building block

2 - AWS SNS/SQS

Component format

To set up AWS SNS/SQS pub/sub, create a component of type pubsub.aws.snssqs.

By default, the AWS SNS/SQS component:

• Generates the SNS topics
• Provisions the SQS queues
• Configures a subscription of the queues to the topics

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: snssqs-pubsub
spec:
  type: pubsub.aws.snssqs
  version: v1
  metadata:
    - name: accessKey
      value: "AKIAIOSFODNN7EXAMPLE"
    - name: secretKey
      value: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
    - name: region
      value: "us-east-1"
    # - name: consumerID # Optional. If not supplied, runtime will create one.
    #   value: "channel1"
    # - name: endpoint # Optional. 
    #   value: "http://localhost:4566"
    # - name: sessionToken  # Optional (mandatory if using AssignedRole; for example, temporary accessKey and secretKey)
    #   value: "TOKEN"
    # - name: messageVisibilityTimeout # Optional
    #   value: 10
    # - name: messageRetryLimit # Optional
    #   value: 10
    # - name: messageReceiveLimit # Optional
    #   value: 10
    # - name: sqsDeadLettersQueueName # Optional
    # - value: "myapp-dlq"
    # - name: messageWaitTimeSeconds # Optional
    #   value: 1
    # - name: messageMaxNumber # Optional
    #   value: 10
    # - name: fifo # Optional
    #   value: "true"
    # - name: fifoMessageGroupID # Optional
    #   value: "app1-mgi"
    # - name: disableEntityManagement # Optional
    #   value: "false"
    # - name: disableDeleteOnRetryLimit # Optional
    #   value: "false"
    # - name: assetsManagementTimeoutSeconds # Optional
    #   value: 5
    # - name: concurrencyMode # Optional
    #   value: "single"
    # - name: concurrencyLimit # Optional
    #   value: "0"

Spec metadata fields

Additional info

Conforming with AWS specifications

Dapr created SNS topic and SQS queue names conform with AWS specifications. By default, Dapr creates an SQS queue name based on the consumer app-id, therefore Dapr might perform name standardization to meet with AWS specifications.

SNS/SQS component behavior

When the pub/sub SNS/SQS component provisions SNS topics, the SQS queues and the subscription behave differently in situations where the component is operating on behalf of a message producer (with no subscriber app deployed), than in situations where a subscriber app is present (with no publisher deployed).

Due to how SNS works without SQS subscription in publisher only setup, the SQS queues and the subscription behave as a “classic” pub/sub system that relies on subscribers listening to topic messages. Without those subscribers, messages:

• Cannot be passed onwards and are effectively dropped
• Are not available for future subscribers (no replay of message when the subscriber finally subscribes)

SQS FIFO

Using SQS FIFO (fifo metadata field set to "true") per AWS specifications provides message ordering and deduplication, but incurs a lower SQS processing throughput, among other caveats.

Specifying fifoMessageGroupID limits the number of concurrent consumers of the FIFO queue used to only one but guarantees global ordering of messages published by the app’s Dapr sidecars. See this AWS blog post to better understand the topic of Message Group IDs and FIFO queues.

To avoid losing the order of messages delivered to consumers, the FIFO configuration for the SQS Component requires the concurrencyMode metadata field set to "single".

Default parallel concurrencyMode

Since v1.8.0, the component supports the "parallel" concurrencyMode as its default mode. In prior versions, the component default behavior was calling the subscriber a single message at a time and waiting for its response.

SQS dead-letter Queues

When configuring the PubSub component with SQS dead-letter queues, the metadata fields messageReceiveLimit and sqsDeadLettersQueueName must both be set to a value. For messageReceiveLimit, the value must be greater than 0 and the sqsDeadLettersQueueName must not be empty string.

SNS/SQS Contention with Dapr

Fundamentally, SNS aggregates messages from multiple publisher topics into a single SQS queue by creating SQS subscriptions to those topics. As a subscriber, the SNS/SQS pub/sub component consumes messages from that sole SQS queue.

However, like any SQS consumer, the component cannot selectively retrieve the messages published to the SNS topics to which it is specifically subscribed. This can result in the component receiving messages originating from topics without associated handlers. Typically, this occurs during:

• Component initialization: If infrastructure subscriptions are ready before component subscription handlers, or
• Shutdown: If component handlers are removed before infrastructure subscriptions.

Since this issue affects any SQS consumer of multiple SNS topics, the component cannot prevent consuming messages from topics lacking handlers. When this happens, the component logs an error indicating such messages were erroneously retrieved.

In these situations, the unhandled messages would reappear in SQS with their receive count decremented after each pull. Thus, there is a risk that an unhandled message could exceed its messageReceiveLimit and be lost.

Create an SNS/SQS instance

• Self-Hosted
• Kubernetes
• AWS

docker run --rm -it -p 4566:4566 -p 4571:4571 -e SERVICES="sts,sns,sqs" -e AWS_DEFAULT_REGION="us-east-1" localstack/localstack

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: snssqs-pubsub
spec:
  type: pubsub.aws.snssqs
  version: v1
  metadata:
    - name: accessKey
      value: "anyString"
    - name: secretKey
      value: "anyString"
    - name: endpoint
      value: http://localhost:4566
    # Use us-east-1 or any other region if provided to localstack as defined by "AWS_DEFAULT_REGION" envvar
    - name: region
      value: us-east-1

apiVersion: apps/v1
kind: Deployment
metadata:
  name: localstack
spec:
  # using the selector, we will expose the running deployments
  # this is how Kubernetes knows, that a given service belongs to a deployment
  selector:
    matchLabels:
      app: localstack
  replicas: 1
  template:
    metadata:
      labels:
        app: localstack
    spec:
      containers:
      - name: localstack
        image: localstack/localstack:latest
        ports:
          # Expose the edge endpoint
          - containerPort: 4566
---
kind: Service
apiVersion: v1
metadata:
  name: localstack
  labels:
    app: localstack
spec:
  selector:
    app: localstack
  ports:
  - protocol: TCP
    port: 4566
    targetPort: 4566
  type: LoadBalancer

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "YOUR_POLICY_NAME",
      "Effect": "Allow",
      "Action": [
        "sns:CreateTopic",
        "sns:GetTopicAttributes",
        "sns:ListSubscriptionsByTopic",
        "sns:Publish",
        "sns:Subscribe",
        "sns:TagResource",
        "sqs:ChangeMessageVisibility",
        "sqs:CreateQueue",
        "sqs:DeleteMessage",
        "sqs:GetQueueAttributes",
        "sqs:GetQueueUrl",
        "sqs:ReceiveMessage",
        "sqs:SetQueueAttributes",
        "sqs:TagQueue"
      ],
      "Resource": [
        "arn:aws:sns:AWS_REGION:AWS_ACCOUNT_ID:*",
        "arn:aws:sqs:AWS_REGION:AWS_ACCOUNT_ID:*"
      ]
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "YOUR_POLICY_NAME",
      "Effect": "Allow",
      "Action": [
        "sqs:DeleteMessage",
        "sqs:ReceiveMessage",
        "sqs:ChangeMessageVisibility",
        "sqs:GetQueueUrl",
        "sqs:GetQueueAttributes",
        "sns:Publish",
        "sns:ListSubscriptionsByTopic",
        "sns:GetTopicAttributes"

      ],
      "Resource": [
        "arn:aws:sns:AWS_REGION:AWS_ACCOUNT_ID:APP_TOPIC_NAME",
        "arn:aws:sqs:AWS_REGION:AWS_ACCOUNT_ID:APP_ID"
      ]
    }
  ]
}

Related links

• Basic schema for a Dapr component
• Pub/Sub building block overview and how-to guides
• Authenticating to AWS
• AWS docs:
  • AWS SQS as subscriber to SNS
  • AWS SNS API reference
  • AWS SQS API reference

3 - Azure Event Hubs

Component format

To set up an Azure Event Hubs pub/sub, create a component of type pubsub.azure.eventhubs. See the pub/sub broker component file to learn how ConsumerID is automatically generated. Read the How-to: Publish and Subscribe guide on how to create and apply a pub/sub configuration.

Apart from the configuration metadata fields shown below, Azure Event Hubs also supports Azure Authentication mechanisms.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: eventhubs-pubsub
spec:
  type: pubsub.azure.eventhubs
  version: v1
  metadata:
    # Either connectionString or eventHubNamespace is required
    # Use connectionString when *not* using Microsoft Entra ID
    - name: connectionString
      value: "Endpoint=sb://{EventHubNamespace}.servicebus.windows.net/;SharedAccessKeyName={PolicyName};SharedAccessKey={Key};EntityPath={EventHub}"
    # Use eventHubNamespace when using Microsoft Entra ID
    - name: eventHubNamespace
      value: "namespace"
    - name: consumerID # Optional. If not supplied, the runtime will create one.
      value: "channel1"
    - name: enableEntityManagement
      value: "false"
    - name: enableInOrderMessageDelivery
      value: "false"
    # The following four properties are needed only if enableEntityManagement is set to true
    - name: resourceGroupName
      value: "test-rg"
    - name: subscriptionID
      value: "value of Azure subscription ID"
    - name: partitionCount
      value: "1"
    - name: messageRetentionInDays
      value: "3"
    # Checkpoint store attributes
    - name: storageAccountName
      value: "myeventhubstorage"
    - name: storageAccountKey
      value: "112233445566778899"
    - name: storageContainerName
      value: "myeventhubstoragecontainer"
    # Alternative to passing storageAccountKey
    - name: storageConnectionString
      value: "DefaultEndpointsProtocol=https;AccountName=<account>;AccountKey=<account-key>"

Spec metadata fields

Microsoft Entra ID authentication

The Azure Event Hubs pub/sub component supports authentication using all Microsoft Entra ID mechanisms. For further information and the relevant component metadata fields to provide depending on the choice of Microsoft Entra ID authentication mechanism, see the docs for authenticating to Azure.

Example Configuration

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: eventhubs-pubsub
spec:
  type: pubsub.azure.eventhubs
  version: v1
  metadata:
    # Azure Authentication Used
    - name: azureTenantId
      value: "***"
    - name: azureClientId
      value: "***"
    - name: azureClientSecret
      value: "***"
    - name: eventHubNamespace 
      value: "namespace"
    - name: enableEntityManagement
      value: "false"
    # The following four properties are needed only if enableEntityManagement is set to true
    - name: resourceGroupName
      value: "test-rg"
    - name: subscriptionID
      value: "value of Azure subscription ID"
    - name: partitionCount
      value: "1"
    - name: messageRetentionInDays
    # Checkpoint store attributes
    # In this case, we're using Microsoft Entra ID to access the storage account too
    - name: storageAccountName
      value: "myeventhubstorage"
    - name: storageContainerName
      value: "myeventhubstoragecontainer"

Sending and receiving multiple messages

Azure Eventhubs supports sending and receiving multiple messages in a single operation using the bulk pub/sub API.

Configuring bulk publish

To set the metadata for bulk publish operation, set the query parameters on the HTTP request or the gRPC metadata, as documented in the API reference.

Configuring bulk subscribe

When subscribing to a topic, you can configure bulkSubscribe options. Refer to Subscribing messages in bulk for more details and to learn more about the bulk subscribe API.

Configuring checkpoint frequency

When subscribing to a topic, you can configure the checkpointing frequency in a partition by setting the metadata in the HTTP or gRPC subscribe request . This metadata enables checkpointing after the configured number of events within a partition event sequence. Disable checkpointing by setting the frequency to 0.

Learn more about checkpointing.

Following example shows a sample subscription file for Declarative subscription using checkPointFrequencyPerPartition metadata. Similarly, you can also pass the metadata in Programmatic subscriptions as well.

apiVersion: dapr.io/v2alpha1
kind: Subscription
metadata:
  name: order-pub-sub
spec:
  topic: orders
  routes: 
    default: /checkout
  pubsubname: order-pub-sub
  metadata:
    checkPointFrequencyPerPartition: 1
scopes:
- orderprocessing
- checkout

Create an Azure Event Hub

Follow the instructions on the documentation to set up Azure Event Hubs.

Because this component uses Azure Storage as checkpoint store, you will also need an Azure Storage Account. Follow the instructions on the documentation to manage the storage account access keys.

See the documentation on how to get the Event Hubs connection string (note this is not for the Event Hubs namespace).

Create consumer groups for each subscriber

For every Dapr app that wants to subscribe to events, create an Event Hubs consumer group with the name of the Dapr app ID. For example, a Dapr app running on Kubernetes with dapr.io/app-id: "myapp" will need an Event Hubs consumer group named myapp.

Note: Dapr passes the name of the consumer group to the Event Hub, so this is not supplied in the metadata.

Entity Management

When entity management is enabled in the metadata, as long as the application has the right role and permissions to manipulate the Event Hub namespace, Dapr can automatically create the Event Hub and consumer group for you.

The Evet Hub name is the topic field in the incoming request to publish or subscribe to, while the consumer group name is the name of the Dapr app which subscribes to a given Event Hub. For example, a Dapr app running on Kubernetes with name dapr.io/app-id: "myapp" requires an Event Hubs consumer group named myapp.

Entity management is only possible when using Microsoft Entra ID Authentication and not using a connection string.

Dapr passes the name of the consumer group to the Event Hub, so this is not supplied in the metadata.

Receiving custom properties

By default, Dapr does not forward custom properties. However, by setting the subscription metadata requireAllProperties to "true", you can receive custom properties as HTTP headers.

apiVersion: dapr.io/v2alpha1
kind: Subscription
metadata:
  name: order-pub-sub
spec:
  topic: orders
  routes: 
    default: /checkout
  pubsubname: order-pub-sub
  metadata:
    requireAllProperties: "true"

The same can be achieved using the Dapr SDK:

• .NET

[Topic("order-pub-sub", "orders")]
[TopicMetadata("requireAllProperties", "true")]
[HttpPost("checkout")]
public ActionResult Checkout(Order order, [FromHeader] int priority)
{
    return Ok();
}

Subscribing to Azure IoT Hub Events

Azure IoT Hub provides an endpoint that is compatible with Event Hubs, so the Azure Event Hubs pubsub component can also be used to subscribe to Azure IoT Hub events.

The device-to-cloud events created by Azure IoT Hub devices will contain additional IoT Hub System Properties, and the Azure Event Hubs pubsub component for Dapr will return the following as part of the response metadata:

For example, the headers of a delivered HTTP subscription message would contain:

{
  'user-agent': 'fasthttp',
  'host': '127.0.0.1:3000',
  'content-type': 'application/json',
  'content-length': '120',
  'iothub-connection-device-id': 'my-test-device',
  'iothub-connection-auth-generation-id': '637618061680407492',
  'iothub-connection-auth-method': '{"scope":"module","type":"sas","issuer":"iothub","acceptingIpFilterRule":null}',
  'iothub-connection-module-id': 'my-test-module-a',
  'iothub-enqueuedtime': '2021-07-13T22:08:09Z',
  'message-id': 'my-custom-message-id',
  'x-opt-sequence-number': '35',
  'x-opt-enqueued-time': '2021-07-13T22:08:09Z',
  'x-opt-offset': '21560',
  'traceparent': '00-4655608164bc48b985b42d39865f3834-ed6cf3697c86e7bd-01'
}

Related links

• Basic schema for a Dapr component
• Read this guide for instructions on configuring pub/sub components
• Pub/Sub building block
• Authentication to Azure

4 - Azure Service Bus Queues

Component format

To set up Azure Service Bus Queues pub/sub, create a component of type pubsub.azure.servicebus.queues. See the pub/sub broker component file to learn how ConsumerID is automatically generated. Read the How-to: Publish and Subscribe guide on how to create and apply a pub/sub configuration.

This component uses queues on Azure Service Bus; see the official documentation for the differences between topics and queues. For using topics, see the Azure Service Bus Topics pubsub component.

Connection String Authentication

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: servicebus-pubsub
spec:
  type: pubsub.azure.servicebus.queues
  version: v1
  metadata:
  # Required when not using Microsoft Entra ID Authentication
  - name: connectionString
    value: "Endpoint=sb://{ServiceBusNamespace}.servicebus.windows.net/;SharedAccessKeyName={PolicyName};SharedAccessKey={Key};EntityPath={ServiceBus}"
  # - name: consumerID # Optional
  #   value: channel1
  # - name: timeoutInSec # Optional
  #   value: 60
  # - name: handlerTimeoutInSec # Optional
  #   value: 60
  # - name: disableEntityManagement # Optional
  #   value: "false"
  # - name: maxDeliveryCount # Optional
  #   value: 3
  # - name: lockDurationInSec # Optional
  #   value: 60
  # - name: lockRenewalInSec # Optional
  #   value: 20
  # - name: maxActiveMessages # Optional
  #   value: 10000
  # - name: maxConcurrentHandlers # Optional
  #   value: 10
  # - name: defaultMessageTimeToLiveInSec # Optional
  #   value: 10
  # - name: autoDeleteOnIdleInSec # Optional
  #   value: 3600
  # - name: minConnectionRecoveryInSec # Optional
  #   value: 2
  # - name: maxConnectionRecoveryInSec # Optional
  #   value: 300
  # - name: maxRetriableErrorsPerSec # Optional
  #   value: 10
  # - name: publishMaxRetries # Optional
  #   value: 5
  # - name: publishInitialRetryIntervalInMs # Optional
  #   value: 500

Spec metadata fields

Microsoft Entra ID authentication

The Azure Service Bus Queues pubsub component supports authentication using all Microsoft Entra ID mechanisms, including Managed Identities. For further information and the relevant component metadata fields to provide depending on the choice of Microsoft Entra ID authentication mechanism, see the docs for authenticating to Azure.

Example Configuration

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: servicebus-pubsub
spec:
  type: pubsub.azure.servicebus.queues
  version: v1
  metadata:
  - name: namespaceName
    # Required when using Azure Authentication.
    # Must be a fully-qualified domain name
    value: "servicebusnamespace.servicebus.windows.net"
  - name: azureTenantId
    value: "***"
  - name: azureClientId
    value: "***"
  - name: azureClientSecret
    value: "***"

Message metadata

Azure Service Bus messages extend the Dapr message format with additional contextual metadata. Some metadata fields are set by Azure Service Bus itself (read-only) and others can be set by the client when publishing a message.

Sending a message with metadata

To set Azure Service Bus metadata when sending a message, set the query parameters on the HTTP request or the gRPC metadata as documented here.

• metadata.MessageId
• metadata.CorrelationId
• metadata.SessionId
• metadata.Label
• metadata.ReplyTo
• metadata.PartitionKey
• metadata.To
• metadata.ContentType
• metadata.ScheduledEnqueueTimeUtc
• metadata.ReplyToSessionId

Receiving a message with metadata

When Dapr calls your application, it attaches Azure Service Bus message metadata to the request using either HTTP headers or gRPC metadata. In addition to the settable metadata listed above, you can also access the following read-only message metadata.

• metadata.DeliveryCount
• metadata.LockedUntilUtc
• metadata.LockToken
• metadata.EnqueuedTimeUtc
• metadata.SequenceNumber

To find out more details on the purpose of any of these metadata properties refer to the official Azure Service Bus documentation.

In addition, all entries of ApplicationProperties from the original Azure Service Bus message are appended as metadata.<application property's name>.

Sending and receiving multiple messages

Azure Service Bus supports sending and receiving multiple messages in a single operation using the bulk pub/sub API.

Configuring bulk publish

To set the metadata for bulk publish operation, set the query parameters on the HTTP request or the gRPC metadata as documented here

Configuring bulk subscribe

When subscribing to a topic, you can configure bulkSubscribe options. Refer to Subscribing messages in bulk for more details. Learn more about the bulk subscribe API.

Create an Azure Service Bus broker for queues

Follow the instructions here on setting up Azure Service Bus Queues.

Retry policy and dead-letter queues

By default, an Azure Service Bus Queue has a dead-letter queue. The messages are retried the amount given for maxDeliveryCount. The default maxDeliveryCount value defaults to 10, but can be set up to 2000. These retries happen very rapidly and the message is put in the dead-letter queue if no success is returned.

Dapr Pub/sub offers its own dead-letter queue concept that lets you control the retry policy and subscribe to the dead-letter queue through Dapr.

1. Set up a separate queue as that dead-letter queue in the Azure Service Bus namespace, and a resilience policy that defines how to retry.
2. Subscribe to the topic to get the failed messages and deal with them.

For example, setting up a dead-letter queue orders-dlq in the subscription and a resiliency policy lets you subscribe to the topic orders-dlq to handle failed messages.

For more details on setting up dead-letter queues, see the dead-letter article.

Related links

• Basic schema for a Dapr component
• Pub/Sub building block
• Read this guide for instructions on configuring pub/sub components

5 - Azure Service Bus Topics

Component format

To set up Azure Service Bus Topics pub/sub, create a component of type pubsub.azure.servicebus.topics. See the pub/sub broker component file to learn how ConsumerID is automatically generated. Read the How-to: Publish and Subscribe guide on how to create and apply a pub/sub configuration.

This component uses topics on Azure Service Bus; see the official documentation for the differences between topics and queues.
For using queues, see the Azure Service Bus Queues pubsub component.

Connection String Authentication

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: servicebus-pubsub
spec:
  type: pubsub.azure.servicebus.topics
  version: v1
  metadata:
  # Required when not using Microsoft Entra ID Authentication
  - name: connectionString
    value: "Endpoint=sb://{ServiceBusNamespace}.servicebus.windows.net/;SharedAccessKeyName={PolicyName};SharedAccessKey={Key};EntityPath={ServiceBus}"
  # - name: consumerID # Optional: defaults to the app's own ID
  #   value: channel1 
  # - name: timeoutInSec # Optional
  #   value: 60
  # - name: handlerTimeoutInSec # Optional
  #   value: 60
  # - name: disableEntityManagement # Optional
  #   value: "false"
  # - name: maxDeliveryCount # Optional
  #   value: 3
  # - name: lockDurationInSec # Optional
  #   value: 60
  # - name: lockRenewalInSec # Optional
  #   value: 20
  # - name: maxActiveMessages # Optional
  #   value: 10000
  # - name: maxConcurrentHandlers # Optional
  #   value: 10
  # - name: defaultMessageTimeToLiveInSec # Optional
  #   value: 10
  # - name: autoDeleteOnIdleInSec # Optional
  #   value: 3600
  # - name: minConnectionRecoveryInSec # Optional
  #   value: 2
  # - name: maxConnectionRecoveryInSec # Optional
  #   value: 300
  # - name: maxRetriableErrorsPerSec # Optional
  #   value: 10
  # - name: publishMaxRetries # Optional
  #   value: 5
  # - name: publishInitialRetryIntervalInMs # Optional
  #   value: 500

NOTE: The above settings are shared across all topics that use this component.

Spec metadata fields

Microsoft Entra ID authentication

The Azure Service Bus Topics pubsub component supports authentication using all Microsoft Entra ID mechanisms, including Managed Identities. For further information and the relevant component metadata fields to provide depending on the choice of Microsoft Entra ID authentication mechanism, see the docs for authenticating to Azure.

Example Configuration

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: servicebus-pubsub
spec:
  type: pubsub.azure.servicebus.topics
  version: v1
  metadata:
  - name: namespaceName
    # Required when using Azure Authentication.
    # Must be a fully-qualified domain name
    value: "servicebusnamespace.servicebus.windows.net"
  - name: azureTenantId
    value: "***"
  - name: azureClientId
    value: "***"
  - name: azureClientSecret
    value: "***"

Message metadata

Azure Service Bus messages extend the Dapr message format with additional contextual metadata. Some metadata fields are set by Azure Service Bus itself (read-only) and others can be set by the client when publishing a message.

Sending a message with metadata

To set Azure Service Bus metadata when sending a message, set the query parameters on the HTTP request or the gRPC metadata as documented here.

• metadata.MessageId
• metadata.CorrelationId
• metadata.SessionId
• metadata.Label
• metadata.ReplyTo
• metadata.PartitionKey
• metadata.To
• metadata.ContentType
• metadata.ScheduledEnqueueTimeUtc
• metadata.ReplyToSessionId

Note: The metadata.MessageId property does not set the id property of the cloud event returned by Dapr and should be treated in isolation.

NOTE: If the metadata.SessionId property is not set but the topic requires sessions then an empty session id will be used.

NOTE: The metadata.ScheduledEnqueueTimeUtc property supports the RFC1123 and RFC3339 timestamp formats.

Receiving a message with metadata

When Dapr calls your application, it will attach Azure Service Bus message metadata to the request using either HTTP headers or gRPC metadata. In addition to the settable metadata listed above, you can also access the following read-only message metadata.

• metadata.DeliveryCount
• metadata.LockedUntilUtc
• metadata.LockToken
• metadata.EnqueuedTimeUtc
• metadata.SequenceNumber

To find out more details on the purpose of any of these metadata properties refer to the official Azure Service Bus documentation.

In addition, all entries of ApplicationProperties from the original Azure Service Bus message are appended as metadata.<application property's name>.

Note: that all times are populated by the server and are not adjusted for clock skews.

Subscribe to a session enabled topic

To subscribe to a topic that has sessions enabled you can provide the following properties in the subscription metadata.

• requireSessions (default: false)
• sessionIdleTimeoutInSec (default: 60)
• maxConcurrentSessions (default: 8)

Create an Azure Service Bus broker for topics

Follow the instructions here on setting up Azure Service Bus Topics.

Related links

• Basic schema for a Dapr component
• Pub/Sub building block
• Read this guide for instructions on configuring pub/sub components

6 - GCP

Create a Dapr component

To set up GCP pub/sub, create a component of type pubsub.gcp.pubsub. See the pub/sub broker component file to learn how ConsumerID is automatically generated. Read the How-to: Publish and Subscribe guide on how to create and apply a pub/sub configuration.

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: gcp-pubsub
spec:
  type: pubsub.gcp.pubsub
  version: v1
  metadata:
  - name: type
    value: service_account
  - name: projectId
    value: <PROJECT_ID> # replace
  - name: endpoint # Optional.
    value: "http://localhost:8085"
  - name: consumerID # Optional - defaults to the app's own ID
    value: <CONSUMER_ID>
  - name: identityProjectId
    value: <IDENTITY_PROJECT_ID> # replace
  - name: privateKeyId
    value: <PRIVATE_KEY_ID> #replace
  - name: clientEmail
    value: <CLIENT_EMAIL> #replace
  - name: clientId
    value: <CLIENT_ID> # replace
  - name: authUri
    value: https://accounts.google.com/o/oauth2/auth
  - name: tokenUri
    value: https://oauth2.googleapis.com/token
  - name: authProviderX509CertUrl
    value: https://www.googleapis.com/oauth2/v1/certs
  - name: clientX509CertUrl
    value: https://www.googleapis.com/robot/v1/metadata/x509/<PROJECT_NAME>.iam.gserviceaccount.com #replace PROJECT_NAME
  - name: privateKey
    value: <PRIVATE_KEY> # replace x509 cert
  - name: disableEntityManagement
    value: "false"
  - name: enableMessageOrdering
    value: "false"
  - name: orderingKey # Optional
    value: <ORDERING_KEY>
  - name: maxReconnectionAttempts # Optional
    value: 30
  - name: connectionRecoveryInSec # Optional
    value: 2
  - name: deadLetterTopic # Optional
    value: <EXISTING_PUBSUB_TOPIC>
  - name: maxDeliveryAttempts # Optional
    value: 5
  - name: maxOutstandingMessages # Optional
    value: 1000
  - name: maxOutstandingBytes # Optional
    value: 1000000000
  - name: maxConcurrentConnections # Optional
    value: 10

Spec metadata fields