dapr / dapr/components-contrib

Propagate headers as properties in Pulsar

Open
#2,819 3 comments 1 reaction 0 assignees View on GitHub
help wanted kind/enhancement P2 pinned
Dominant language
Go
Stars
602
Forks
580
Avg merge
4d 9h
Merged PRs (30d)
6

Description

### Description
Propagate all headers as properties in Pulsar pubsub components just like its done in Kafka ( see: https://github.com/dapr/components-contrib/pull/2017 ).

The business value would be related to this long standing feature request: https://github.com/dapr/dapr/issues/6075 where users can keep metadata (aka headers/properties/etc) outside of the CloudEvent Dapr sends.

### Additional details
When using Kafka as my underlying pubsub system I am able to send headers in ( see: https://docs.dapr.io/reference/components-reference/supported-pubsub/setup-apache-kafka/#message-headers ) which is the behavior I wanted to see in Pulsar.

To give some more context on how this was accomplished I put all headers into the PublishEventRequest.metadata map:
...
for (Map.Entry header : headers.entrySet()) {
metadata.put(header.getKey(), header.getValue());
}
publishEventRequest.setMetadata(metadata);
...

My sample command has an additional "metadata.correlationId" header.
$ curl --request POST http://localhost:8080/pub -H "Content-Type: application/json" -H "cloudevent.type: typeGoesHere" -H "metadata.correlationId: correlationIdGoesHere" -d '{"id":1,"name":"Bob"}'

In Kafka we can see that the "headers" are stored as "headers":
Topic orders-with-schema2[0], offset: 3, key: , payload: 444, bytes: {"data":"{"id":1,"name":"Bob"}","datacontenttype":"application/json","id":"fba1f6e2-23a7-40a0-9b79-c7d9d023191b","pubsubname":"evmg-pubsub-component","source":"evmg-dapr-poc","specversion":"1.0","time":"2023-04-27T15:48:29-07:00","topic":"orders-with-schema2","traceid":"00-64c50d00d78178347c2202c5caf1f36e-3a50cb7548c163cb-01","traceparent":"00-64c50d00d78178347c2202c5caf1f36e-3a50cb7548c163cb-01","tracestate":"","type":"typeGoesHere"}, headers: metadata.correlationid=correlationIdGoesHere,user-agent=curl/7.87.0,accept=/,content-length=21,host=localhost:8080,cloudevent.type=typeGoesHere,content-type=application/json

In Dapr I can also see the Kafka "headers" are propagated through as "headers":
== APP == [2023-04-27 16:41:00.202] DaprPOC [http-nio-8080-exec-3] INFO c.f.e.d.p.p.DaprPOCPub.67: Publishing data: "{"id":1,"name":"Bob"}", metadata: "{content-length=21, host=localhost:8080, cloudevent.type=typeGoesHere, content-type=application/json, metadata.correlationid=correlationIdGoesHere, user-agent=curl/7.87.0, accept=/}", contentType: "null" to topicName: "orders-with-schema2" on pubsubName: "evmg-pubsub-component"
== APP == [2023-04-27 16:41:00.638] DaprPOC [http-nio-8080-exec-3] INFO c.f.e.d.p.p.DaprPOCPub.69: Published data: "{"id":1,"name":"Bob"}", metadata: "{content-length=21, host=localhost:8080, cloudevent.type=typeGoesHere, content-type=application/json, metadata.correlationid=correlationIdGoesHere, user-agent=curl/7.87.0, accept=/}", contentType: "null" to topicName: "orders-with-schema2" on pubsubName: "evmg-pubsub-component"
== APP == [2023-04-27 16:41:00.679] DaprPOC [http-nio-8080-exec-4] INFO c.f.e.d.p.s.DaprPOCSub.19: Received cloudEvent with data: "{"id":1,"name":"Bob"}", binaryData (decoded): "null", id: "edaed576-6c0c-4278-a66c-be61d5d82557", source: "evmg-dapr-poc", type: "typeGoesHere", dataContentType: "application/json", specVersion: "1.0", and headers: "{host=127.0.0.1:8080, user-agent=curl/7.87.0, content-length=444, accept=/, cloudevent.type=typeGoesHere, content-type=application/cloudevents+json, metadata.correlationid=correlationIdGoesHere, pubsubname=evmg-pubsub-component, traceparent=00-6461f30a2d15ef1bd695f427bd581d5c-65fff60a8e55703b-01, accept-encoding=gzip}"

On the other hand, in Pulsar we can see that no "headers" are stored as "properties":
----- got message -----
key:[null], properties:[], content:{"data":"{"id":1,"name":"Bob"}","datacontenttype":"application/json","id":"1809c398-75e0-489e-9cc4-0945fcd7a498","pubsubname":"evmg-pubsub-component","source":"evmg-dapr-poc","specversion":"1.0","time":"2023-04-27T16:49:23-07:00","topic":"orders-with-schema2","traceid":"00-2146a01921907ccb2d1891a92e7cc42d-84aa1d5c2c93ab4f-01","traceparent":"00-2146a01921907ccb2d1891a92e7cc42d-84aa1d5c2c93ab4f-01","tracestate":"","type":"typeGoesHere"}

Thus when I check Dapr I don't see any Pulsar "properties" propagated through as "headers":
== APP == [2023-04-27 16:44:02.752] DaprPOC [http-nio-8080-exec-3] INFO c.f.e.d.p.p.DaprPOCPub.67: Publishing data: "{"id":1,"name":"Bob"}", metadata: "{content-length=21, host=localhost:8080, cloudevent.type=typeGoesHere, content-type=application/json, metadata.correlationid=correlationIdGoesHere, user-agent=curl/7.87.0, accept=/}", contentType: "null" to topicName: "orders-with-schema2" on pubsubName: "evmg-pubsub-component"
== APP == [2023-04-27 16:44:03.363] DaprPOC [http-nio-8080-exec-3] INFO c.f.e.d.p.p.DaprPOCPub.69: Published data: "{"id":1,"name":"Bob"}", metadata: "{content-length=21, host=localhost:8080, cloudevent.type=typeGoesHere, content-type=application/json, metadata.correlationid=correlationIdGoesHere, user-agent=curl/7.87.0, accept=/}", contentType: "null" to topicName: "orders-with-schema2" on pubsubName: "evmg-pubsub-component"
== APP == [2023-04-27 16:44:03.402] DaprPOC [http-nio-8080-exec-4] INFO c.f.e.d.p.s.DaprPOCSub.19: Received cloudEvent with data: "{"id":1,"name":"Bob"}", binaryData (decoded): "null", id: "c3ebec99-d86a-4e1b-9c5a-8fc0c4f94df0", source: "evmg-dapr-poc", type: "typeGoesHere", dataContentType: "application/json", specVersion: "1.0", and headers: "{host=127.0.0.1:8080, user-agent=Go-http-client/1.1, content-length=444, content-type=application/cloudevents+json, pubsubname=evmg-pubsub-component, traceparent=00-96a84f878a3955af03ae8cf3a882689f-f020885e7f503cd5-01, accept-encoding=gzip}"

So this appears to be something that can indeed be added to the Pulsar pubsub component and others as outlined by this excellent comment by @mukundansundar ( https://github.com/dapr/dapr/issues/6075#issuecomment-1468902729 ).

### Release note
**ADD** Propagation of headers as properties in Pulsar pubsub components

Contributor guide

Open the contributing guide

Research direction

Locate the Pulsar pubsub component and compare its publish and subscribe paths with the Kafka header handling described in pull request 2017. Verify that request metadata is stored as Pulsar properties and is available as headers when consumed; add or run component tests covering the metadata.correlationId example.

Written by the indexing model from the issue text.

Assessment

Tech stack
go
Domain
backend, distributed-systems
Issue type
Feature
Difficulty
3/5
Estimated time
1-2 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.