Proposal

Overview

The current PubSub API is extremely simple but lacks very important configurations and features. This proposal includes a set of changes to be implemented for the PubSub API.

1. Missing retriable vs non-retriable error code handling

Publishing a message to Dapr can fail but the error might not be retriable. The publishing API only supports two status codes: 200 and 500. There are many error codes that can indicate if an error is not retriable. This is very important, so the client can know that a given error is deterministic and no further retries should be done. For example, a 400 status code should indicate to the client that the request sent to Dapr is not valid. The same happens for the subscribers, when Dapr expects only 200 and 500 as valid status code from client's API.

Expeted behavior: expand the list of expected error codes and specify which errors are retriable vs non-retriable.

Update

Status: DONE in #1993

Retry logic is not exactly how described in this issue. Dapr's current logic to handle error codes is described here https://github.com/dapr/docs/blob/master/reference/api/pubsub_api.md

2. Hard-coding of /<topic> for subscriber APIs

When an applications subscribes to one or more topics, Dapr assumes one / api per topic. This is a strong assumption and it limits the design choices for the customer APIs. The client, might want to have a RESTful compliant API but Dapr would violate that with this enforcement.

Expected behavior: the subscriber response should returns a list of objects (and not a list of strings), so each subscribed topic can have many addition properties: URL endpoint, Retry Timeout, etc.

Example of subscriber's response:

{
  "subscriptions": [
    {
       "topic": "hotels",
       "callback": "/dapr/orders/hotels"
    },
    { 
       "topic": "comments",
       "callback": "/dapr/reviews/comments",
       "retryTimeInMinutes": 60,
       "retryLogic": "exponentialBackoff"
    }
  ]
}

Update

Status: DONE in #1529

Custom route is supported and documented in https://github.com/dapr/docs/blob/master/reference/api/pubsub_api.md. Configuration per topic subscription can be further extended but specific new configs (like retryLogic) is not an ask here, just an example of future possibilities with this new schema.

3. Client-provided message Id

ServiceBus supports client-provided Message Id. The idea is that the publisher can take ownership of Id generation for the messages and take advantage of extra features in the underlying service. In ServiceBus, Message Id can be used to dedup messages during a time interval, providing free one-publish guarantee. Not every implementation will support client-provided Message Ids, so it would be ignored in those cases.

Expedted behavior: publisher can (optionally) provide a special attribute in the payload (e.g. "id" or "dapr.messageId") and Dapr can extract it and use it as the message id (if the component supports it). The message format can also enforce the CloudEvents spec of uniqueness of "source" + "id".

Example of publisher's message:

{
  "id": 123456,
  "data": {
     "order": 42
  }
}

Update

Status: DONE in #1372

Users can accomplish this by publishing a cloud event directly to Dapr. This way, Dapr will not wrap it inside another CloudEvent, keeping the id provided by the user. Publishing the following event, for example, keeps the id on the subscriber side.

    {
        "id" : "100",
        "specversion" : "1.0",
        "type" : "com.github.pull.create",
        "source" : "https://github.com/cloudevents/spec/pull",
        "subject" : "123",
        "time" : "2018-04-05T17:31:00Z",
        "comexampleextension1" : "value",
        "comexampleothervalue" : 5,
        "datacontenttype" : "text/xml",
        "data" : "<much wow=\"xml\"/>"
    }

4. Message Expiration

ServiceBus offers native support for TTL per message. Kafka does it at a global or topic level but not at message level. Dapr can use native support for some implementation but also implement TTL by not publishing messages that have expired. The same will happen if a message cannot be sent to a subscriber after retrying for a configurable amount of time.

Expected behavior: Publisher can (optionally) enter the expiration date/time in UTC as part of a CloudEvent payload (e.g. "daprMessageExpiration") and Dapr will not publish (or retry) the message after the given expiration.

Example of publisher's message:

    {
        "id" : "100",
        "specversion" : "1.0",
        "type" : "com.github.pull.create",
        "source" : "https://github.com/cloudevents/spec/pull",
        "subject" : "123",
        "time" : "2018-04-05T17:31:00Z",
        "daprMessageExpiration" : "2020-04-05T17:31:00Z",
        "comexampleothervalue" : 5,
        "datacontenttype" : "text/xml",
        "data" : "<much wow=\"xml\"/>"
    }

Update

Not done yet. Tracking in #2216

5. DeadLetter Topic

After a message cannot be delivered to a subscriber (retry time out or message expiration), the message should be sent to a special topic. The subscriber can subscribe from that topic as well, or simply leave messages there to be manually verified. Dapr will try to publish to dead letter topic only once. This is a way the subscriber can recover if the API was returning non-retriable errors wrongly due to a code bug, for example, or messages expired after being retried for too long. This allows subscribers to self-recover without escalating to the publisher.

Expected behavior: (1) Subscriber can configure a deadletter topic for a given subscription. (2) Subscribers can subscribe to a deadletter topic like any other topic.

Example of subscriber's response:

{
  "subscriptions": [
    {
       "topic": "hotels",
       "callback": "/dapr/orders/hotels"
    },
    { 
       "topic": "comments",
       "callback": "/dapr/reviews/comments",
       "deadLetterTopic": "comments_myconsumergroup_deadletter",
    },
    { 
       "topic": "comments_myconsumergroup_deadletter",
       "callback": "/dapr/reviews/comments_deadletter",
    }
  ]
}

Update

Not done yet. Tracking in #2217

6. Queue support

The current PubSub model is a broadcast model because one published message will be consumed (ideally) by all subscribers - except when there is a major outage by one of them. SQS, for example, is a queue - it means at least one of the subscribers will receive that message but not necessarily more. Dapr should enable both PubSub and Queue patterns. A Queue should not allow more than one subscriber application. An application can have multiple nodes, so they take care of redundancy under a given application.

Expected behavior: (1) Create a Queue component in Dapr.

Update

Status: ARCHIVED as per #771

This feature can be implemented on each component. PubSub can behave as a topic or queue depending on the implementation. In EventHub, for example, the consumerGroupId can define the behavior of queue vs topic.

7. Batching

The current PubSub model notifies consumers by making a roundtrip per message. A way to improve message throughput would be to allow consumers to declare the intention of consuming messages in batches. Processing messages in batches lead to the following optimizations:

• roundtrip reduction between dapr and consuming component by sending messages batches
• consumer computation by observing a message batch instead of a single item

Expected behavior: Subscribers can (optionally) choose to receive messages in batches.

Example of a subscriber's response where batching is requested:

{
  "subscriptions": [
    {
       "topic": "A",
       "maxBatchSize": "10"
    }
  ]
}

The optional parameter maxBatchSize defines the maximum number of messages to be delivered in a single receive loop. By default, maxBatchSize should be 1.

Message checkpointing in batch mode are for all messages (all succeeded or failed).

Distributed tracing should track each message inside the batch in such a way that users can still find them in the target telemetry backend. In other words, if messages A, B and C were processed in the same batch, searching for message B telemetry should return the batch execution.

In order to enable receiving multiple messages the subscriber callback contract could be changed from:

// NewMessage is an event arriving from a message bus instance
type NewMessage struct {
	Data     []byte            `json:"data"`
	Topic    string            `json:"topic"`
	Metadata map[string]string `json:"metadata"`
}

to

// Message is one event arriving from a message bus instance
type Message struct {
	Data []byte `json:"data"`
	Metadata map[string]string `json:"metadata"`
}

// NewMessages contains one or more events arriving from a message bus instance for a given topic
type NewMessages struct {
	Messages []Message `json:"messages"`
	Topic    string    `json:"topic"`
}

Update

Not done yet. Tracking in #2218

F.A.Q.

1. Should each component implementation handle dead letter queue instead?

Dapr does not stop implementations from handling deadletetter queue or topic, but dapr can offer this feature even to components that do not this feature built in. This deadletter topic goes side by side with message expiration feature.

2. Should Dapr delegate batching to component implementation?

Without this feature, components can still support batching via metadata but the messages will still be sent to subscribers one-by-one. Having this feature built in Dapr (breaking the API) will allow apps to receive messages in actual batches (one request with many events).

Archived

This issue is being closed and archived since we now have individual items tracking each feature.