dapr / dapr/components-contrib
Propagate headers as properties in Pulsar
- 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
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