Spring Pulsar Message Key

[horiaradu1]

Hello

I have a question about producing messages with headers on Pulsar.

I have tried it 2 separate ways, with a PulsarTemplate and a Pulsar Producer Client:

pulsarTemplate.newMessage(message)
                         .withMessageCustomizer((mb) -> mb.key(key))
                         .send();

producer.newMessage()
               .key(key)
               .value(message)
               .send();

Both of them sent the payload correctly, with a header named "pulsar_message_key", which had the value of the key I had given it.

I then tried sending the key, using the binder (I have a setup similar to sample-pulsar-binder).

streamBridge.send(bindingName, MessageBuilder.withPayload(message)
                                                                                .setHeader(PulsarHeaders.KEY, key)
                                                                                .build());

But the "pulsar_message_key" header gets removed by the outBoundMatchers in AbstractPulsarHeaderMapper.java, those internal framework headers being in the NeverMatch list.

Is there a way that I am missing to set up the key (pulsar_message_key) using the binder method, maybe set them up in the binder producer configuration?
Or am I not correctly understanding how those internal headers should be set up and work?

Thank you all for the work you have done on this project, and please let me know if I missed any details.

[onobc]

Hi @horiaradu1

Thank you all for the work you have done on this project,

You are more than welcome. And thank you for using the framework and for opening the discussion.

Good sleuthing on why the headers are not visible in the binder case (the mapper is dropping them). This works in the non-binder scenarios because no mapper is involved.

Typically, you would need to provide a custom header mapper.

The default mapper can be seen here (uses Jackson if available on the classpath otherwise just defaults to ToString mapper).

However, unfortunately, the "never" headers are not currently configurable via the Jackson or ToString constructors. You could create a custom mapper class and override the matchesForOutbound method as follows:

class CustomHeaderMapper extends JsonPulsarHeaderMapper {
	
	CustomHeaderMapper(ObjectMapper objectMapper, List<String> inboundPatterns, List<String> outboundPatterns, Set<String> trustedPackages, Set<String> toStringClasses) {
		super(objectMapper, inboundPatterns, outboundPatterns, trustedPackages, toStringClasses);
	}

	@Override
	protected boolean matchesForOutbound(String header) {
		return PulsarHeaders.KEY.equals(header) || super.matchesForOutbound(header);
	}
}

We will try to get a fix in that allows the "never" headers to be configured via constructors. Hopefully it will be in the 1.2.3 release on 02/18.

Thanks,
Chris

[onobc]

@horiaradu1 I created #1037 to track this. If you are satisfied w/ the above response and the newly created issue I will close this discussion.

Thanks

[horiaradu1]

Hello @onobc

I have done exactly that, created a CustomeHeaderMapper, and allowed the PulsarHeaders.KEY to be mapped to the PulsarHeaders from SpringHeaders, by overriding the matchesForOutbound method. This allowed me to send the desired key value on the pulsar_message_key header.

I have noticed though that with this method, while the property is sent and I can get and parse the value, the key is not recognised as a org.apache.pulsar.client.api.Message 'key', only as a property.
From my consumer tests, I have seen that it does not get into the if (pulsarMessage.hasKey()) if statement from AbstractPulsarHeaderMapper, from toSpringHeaders(Message<?> pulsarMessage) method.

	@Override
	public MessageHeaders toSpringHeaders(Message<?> pulsarMessage) {
		Objects.requireNonNull(pulsarMessage, "pulsarMessage must not be null");
		var context = toSpringHeadersOnStarted(pulsarMessage);
		var headersMap = new LinkedHashMap<String, Object>();

		// custom user properties (headers)
		pulsarMessage.getProperties().forEach((name, value) -> {
			if (matchesForInbound(name)) {
				var valueToUse = toSpringHeaderValue(name, value, context);
				headersMap.put(name, valueToUse);
			}
		});
		// built-in Pulsar metadata headers
		if (pulsarMessage.hasKey()) {
			addToHeadersMapIfAllowed(PulsarHeaders.KEY, pulsarMessage::getKey, headersMap::put);
			addToHeadersMapIfAllowed(PulsarHeaders.KEY_BYTES, pulsarMessage::getKeyBytes, headersMap::put);
		}
		...

What I did instead in the binder case was create a CustomPulsarMessageChannelBinder that extends the PulsarMessageChannelBinder, and rewrote some of the package protected code, to override the protected void handleMessageInternal(Message<?> message) method from PulsarProducerConfigurationMessageHandler.

        @Override
        protected void handleMessageInternal(Message<?> message) {
            try {
                // @formatter:off
                this.pulsarTemplate.newMessage(message.getPayload())
                        .withTopic(this.destination)
                        .withSchema(this.schema)
                        .withProducerCustomizer(this.layeredProducerPropsCustomizer)
                        .withMessageCustomizer(this.applySpringHeadersAsPulsarProperties(message.getHeaders()))
                        // CUSTOM added message customizer for message key
                        .withMessageCustomizer(this.applySpringHeaderAsPulsarKey(message.getHeaders()))
                        //
                        .sendAsync();
                // @formatter:on
            }
            catch (Exception ex) {
                logger.trace(ex, "Failed to send message to destination: " + this.destination);
            }
        }

        private TypedMessageBuilderCustomizer<Object> applySpringHeaderAsPulsarKey(MessageHeaders headers) {
            return (mb) -> {
                if (this.headerMapper != null && headers.containsKey(PulsarHeaders.KEY)) {
                    Optional.ofNullable(headers.get(PulsarHeaders.KEY)).ifPresent(k -> mb.key(k.toString()));
                }
            };
        }
        ...

I added this .withMessageCustomizer(this.applySpringHeaderAsPulsarKey(message.getHeaders())) to the pulsarTemplate send, to map the value I wanted to the message key, instead of message property. This then made the if (pulsarMessage.hasKey()) statment from toSpringHeaders true, and passed both the KEY and KEY_BYTES headers to the consumed message.

Would this also be a good solution to mapping keys to the Pulsar Message, instead of as a property?
And could there be some method/constructor customizability added to a future version of spring-cloud-stream-binder-pulsar that would allow this functionality?

Thank you,
Horia

[onobc]

Hi @horiaradu1 ,

Ahh yes I see the situation. While the KEY is listed as a HEADER, when we reconstitute it as a PROPERTY it looses its special meaning as a KEY.

It seems that the applySpringHeadersAsPulsarProperties could be repurposed to applySpringHeaders and it would do everything except the KEY header as a property and do the KEY header as a KEY.

Have you thought instead of just using a KEY_VALUE schema for the message? Would that solve your use case?

[horiaradu1]

Hello @onobc,

I currently have a large number of Avro schemas in the schema registry with schema validation enforced, which are being used by multiple consumers.
From my understanding, using a KEY_VALUE schema would require modifying the existing schemas, which would impact consumers. Is this correct, or did I misunderstand what you mean?

Having a method like applySpringHeaders that does what you described would work, similar to what I implemented in my workaround. Would it be feasible to add this in a future version of spring-cloud-stream-binder-pulsar?
Additionally, by making this method accessible and customizable, users could extend its functionality to support additional org.apache.pulsar.client.api.TypedMessageBuilder methods, not just for properties, but also for features like orderingKey, sequenceId, etc., without modifying core framework logic.

Perhaps we could even add this applySpringHeaders method to the PulsarHeaderMapper interface, provide a default/simple implementation in AbstractPulsarHeaderMapper, and allow it to be extended via a CustomPulsarHeaderMapper bean. This would, however, require modifying PulsarMessageChannelBinder to use withMessageCustomizer(this.headerMapper.applySpringHeaders(message.getHeaders())) in pulsarTemplate.sendAsync. Would modifying PulsarMessageChannelBinder in this way align with the intended design of the framework?

This is a new technology for me, so I apologize if I am misunderstanding any concepts or missing any key points. I am trying to brainstorm ideas for a possible solution. Any feedback is appreciated.

Thank you very much for your help and time,
Horia

[onobc]

Hi @horiaradu1 ,

I currently have a large number of Avro schemas in the schema registry with schema validation enforced, which are being used by multiple consumers.
From my understanding, using a KEY_VALUE schema would require modifying the existing schemas, which would impact consumers. Is this correct, or did I misunderstand what you mean?

You are correct, this would require you to modify your existing schemas (which can be quite the mess).

Having a method like applySpringHeaders that does what you described would work, similar to what I implemented in my workaround. Would it be feasible to add this in a future version of spring-cloud-stream-binder-pulsar?
Additionally, by making this method accessible and customizable, users could extend its functionality to support additional org.apache.pulsar.client.api.TypedMessageBuilder methods, not just for properties, but also for features like orderingKey, sequenceId, etc., without modifying core framework logic.

I think you have uncovered a weakness here in the current impl that there are some important headers (keys, sequenceId, etc..) that we should reconstitute when converting to Pulsar messages. Currently we are just dropping them on the floor by making them properties.

Perhaps we could even add this applySpringHeaders method to the PulsarHeaderMapper interface, provide a default/simple implementation in AbstractPulsarHeaderMapper, and allow it to be extended via a CustomPulsarHeaderMapper bean. This would, however, require modifying PulsarMessageChannelBinder to use withMessageCustomizer(this.headerMapper.applySpringHeaders(message.getHeaders())) in pulsarTemplate.sendAsync. Would modifying PulsarMessageChannelBinder in this way align with the intended design of the framework?

I am not sure where we should put this customization but I do agree that we should support this. I am inclined to first support "key" header by incuding it on the mapping and then applying it properly w/ the "key" method on the TypedMessageBuilder.

We are currently busy w/ releases this week but I will take some time towards the end of the week to come up w/ a high level proposal wrt to your ideas above. I will reply back here when I do.

This is a new technology for me, so I apologize if I am misunderstanding any concepts or missing any key points. I am trying to brainstorm ideas for a possible solution. Any feedback is appreciated.

No need to apoligize, this is a grea discussion and you have uncovered what I think is a current weakness that needs to be improved. Great sleuthing!

Thank you very much for your help and time,

You are more than welcome. Thank you as well.

Also, you may have already found the docs for the SCS Pulsar binder docs, but they are here and may be helpful to get an overview of the components.

[horiaradu1]

Hello @onobc,

Thank you, I will wait for your update when you have time.
I will take a deeper look at the Spring Cloud Stream and Pulsar Binder docs and source code to gain a better understanding of the framework and components.