solace_replyto header is failed to retrieve in spring cloud stream

as63010

Hi Team,

We are getting the below error when get the solace_replyto header. Please help to check.

Caused by: java.lang.IllegalArgumentException: Incorrect type specified for header 'scst_targetDestination'. Expected [class java.lang.String] but actual type is [class com.solacesystems.jcsmp.impl.TopicImpl]

marc

Hi @as63010,

It looks like you are trying to set the scst_targetDestination header directly to the value of the solace_replyTo header. The solace_replyTo header is actually of type Destination and not String so you have two options:

1. Cast solace_replyTo to a Destination type when reading it and use the getName() method to get the actual String representation of the name OR
2. If you're using all Spring Cloud Stream apps you could use a different (custom) header to capture the topic the requestor wants the replier to respond on and just keep it as a String the whole time.

Hope that helps!

as63010

Thanks Marc for your support

marc

No problem @as63010 - I should have also included that you can find more info on the headers available via the solace binder here: https://github.com/SolaceProducts/solace-spring-cloud/tree/master/solace-spring-cloud-starters/solace-spring-cloud-stream-starter#solace-headers

as63010

Hi @Marc,
When we reply to solace_replyTo topic from spring cloud stream, message not received by requester. Any additional setting is required to reply?

as63010

setHeader(BinderHeaders.TARGET_DESTINATION,response.getHeaders().get("solace_replyTo")).

as63010

@Marc, Can you help this?

We are getting the replyTo topic name as below. Is it possible to put the message in reply topic using spring cloud stream?

solace_replyTo=#P2P/v:solace-xxxx/nDOiG3j8/H17NDCFI13P0120/5084/#00130001/STsW9Glj1L/#

as63010

@JamilAhmed Can you help this? We aren't able to receive the reply from Orchestrator.
We tried the below sample shared in github.

// Build the topics
outputTopic = input.getHeaders().get(MEDIATOR_REPLYTO_DESTINATION_KEY).toString();

Message output = MessageBuilder.withPayload(jsonMessageOut.toString())
.setHeader(SOL_CORRELATION_ID_KEY, input.getHeaders().getOrDefault(MEDIATOR_CORRELATION_ID_KEY, ""))
.setHeader(BinderHeaders.TARGET_DESTINATION, outputTopic)
.build();

JamilAhmed

Hi @as63010, looks like the #P2P... reply topic is correctly being fetched and set as the destination.
In terms of the original request, is that another application using Solace APIs, or are you using the Microgateway feature to automatically convert a HTTP operation into a request message? If using microgateway, did you correctly set the REST mode in your message VPN to gateway? (Docs Link)

as63010

@JamilAhmed Currently we are not using micro gateway. We use other application(developed using native solace JCMS - sendRequest API) to read the reply from orchestrator now. Microgateway will replace this application later.

JamilAhmed

OK let's debug together what is going wrong. I'll dm you to arrange.

marc

Thanks so much @JamilAhmed for jumping in to help out! definitely curious what you find the issue is

JamilAhmed

Hi @marc we confirmed through client statistics that request messages get sent out and a reply is coming back. (Counter increments for each attempts.)

With the original requestor sending a blocking request, it times out and doesn't seem to detect the reply as having come back. We are planning to use the message dump feature of sdkperf to see both directions of the flow and check if all is correct and as expected in terms of correlation IDs and such. We're picking up again Monday.

Will update this thread once resolved.

JamilAhmed

Hi @marc, an update for the thread for those interested...

When an application using jcsmp is the requestor, and sends those requests with the blocking call, any reply messages sent using spring cloud stream will fail to correlate back and a timeout exception is thrown.

For Requestor.request:

BytesXMLMessage request(XMLMessage request,
                        long timeoutMillis,
                        Destination sendDestination)
                 throws JCSMPException

Blocking mode
The timeoutMillis parameter controls whether this call should operate in a blocking fashion.

If >0, the call blocks for up to the number of milliseconds specified by timeoutMillis while the Requestor waits for a reply. If none is received during this time, a JCSMPRequestTimeoutException is thrown.

If 0, the call returns null after the request message has been sent. The application should then examine incoming messages on a XMLMessageConsumer or its XMLMessageListener for a reply, which will be sent to the Destination specified in the request's ReplyTo header field. To correlate replies to sent requests, an application can use the CorrelationId field on outgoing requests.

The issue seems to be the reply messages coming back not having the isReply attribute set on them. If the application was using jcsmp, the XMLProducer.sendReply call takes care of automatically setting this.

Automatically copies requestMessage's CorrelationId to replyMessage and sets the reply attribute on the latter. replyMessage is sent to the Destination returned by requestMessage.getReplyTo().

If sending that message without that convenience function, then you would have called this:
setAsReplyMessage(true)

The only way to make it work is to send the requests non-blocking, and then asychronously handle all the reply messages that arrive in onReceive and do correlation yourself using the value in correlationID.

Since the Solace Binder allows the reply-to destination for a message to be set using the special solace_replyTo message header, there should be a way to allow the isReply attribute to be set on outbound messages too?

Perhaps a new solace_setAsReply header?

marc

Thanks @JamilAhmed and @as63010 for chasing this down. I just wanted to provide an update as I spoke with the RND team our cloud stream binder yesterday. Your troubleshooting is correct that when using the JCSMP Requestor.request convenience function the replier must set a message attribute using the setAsReplyMessage(true) method in JCSMP (or similar methods in other SMF apis). My understanding is that this flag is basically used internally to the broker for enhanced performance when implementing the Req/Reply pattern between SMFs APIs. Unfortunately, the setting of this flag is not currently supported in the Solace Spring Cloud Stream binder.

So I'd say 3 things here:
1. We are going to add support for this in the next release of the Spring Cloud Stream binder. We are planning quite a few things as part of that release and it is being scoped/planned now. With the holiday coming up it is likely that the release w/ this feature will be early in calendar year 2022.
2. Until that feature is added to the Spring Cloud Stream binder, if you are trying to do Req/Reply with JCSMP as the Requestor and Spring Cloud Stream as the Replier you will need to avoid using the JCSMP Requestor.request convenience function and just implement the requestor messaging logic with base JCSMP (set ReplyTo & ApplicationMessageId/CorrelationID headers, listen on the ReplyTo topic, correlate response)
3. Of course you could also write a replier using a different SMF API, but I also want to note that it is currently supported via our JMS API and if you (or anyone else that finds this thread) decide to use JMS as the replier this property can be set using SupportedProperty.Solace_JMS_Prop_IS_Reply_Message

Hope that helps!