Spring cloud stream solace binder StreamBridge.send() unexpected error

siavashsoleymani · July 16, 2024, 1:57pm

This is the github issue I have created:

github.com/SolaceProducts/solace-spring-cloud

[Bug]: Spring cloud stream solace binder StreamBridge.send() unexpected error

opened 09:39AM - 16 Jul 24 UTC

closed 04:07PM - 30 Jul 24 UTC

siaavush

bug

### Bug Description we are using Spring Cloud Stream Solace in our Spring Boot… project, when we try to send messages simultaneously from 5 different threads by using `StreamBridge.send()` the underlying implementation "randomly" throws `org.springframework.messaging.MessagingException`, by randomly I mean it doesn't matter if I try to send 1000 messages at the same time or 50. Library version: `implementation("com.solace.spring.cloud:spring-cloud-starter-stream-solace:4.2.0")` Full stack trace: ``` org.springframework.messaging.MessagingException: Cannot send message using handler 028dfc28-458f-46d9-a669-aa22169323fd at com.solace.spring.cloud.stream.binder.util.ErrorChannelSendingCorrelationKey.send(ErrorChannelSendingCorrelationKey.java:57) at com.solace.spring.cloud.stream.binder.outbound.JCSMPOutboundMessageHandler.handleMessagingException(JCSMPOutboundMessageHandler.java:188) at com.solace.spring.cloud.stream.binder.outbound.JCSMPOutboundMessageHandler.handleMessage(JCSMPOutboundMessageHandler.java:76) at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder$SendingHandler.handleMessageInternal(AbstractMessageChannelBinder.java:1185) at org.springframework.integration.handler.AbstractMessageHandler.doHandleMessage(AbstractMessageHandler.java:105) at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:73) at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:132) at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:133) at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106) at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72) at org.springframework.integration.channel.AbstractMessageChannel.sendInternal(AbstractMessageChannel.java:378) at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:332) at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:302) at org.springframework.cloud.stream.function.StreamBridge.send(StreamBridge.java:187) at org.springframework.cloud.stream.function.StreamBridge.send(StreamBridge.java:146) at org.springframework.cloud.stream.function.StreamBridge.send(StreamBridge.java:141) at com.ingka.fms.inbound.reversereceiving.common.service.EventMeshPublisher.publish(EventMeshPublisher.kt:22) at com.ingka.fms.inbound.reversereceiving.common.service.EventMeshPublisher.publish$default(EventMeshPublisher.kt:18) at com.ingka.fms.inbound.reversereceiving.receiving.messaging.publisher.reverseReceivingUnitStatus.ReverseReceivingUnitStatusUpdatedPublisher.publish(ReverseReceivingUnitStatusUpdatedPublisher.kt:37) at com.ingka.fms.inbound.reversereceiving.receiving.messaging.publisher.reverseReceivingUnitStatus.ReverseReceivingUnitStatusUpdatedPublisher.createEventAndPublish(ReverseReceivingUnitStatusUpdatedPublisher.kt:73) at com.ingka.fms.inbound.reversereceiving.common.service.UnitService.publishEvents(UnitService.kt:281) at com.ingka.fms.inbound.reversereceiving.common.service.UnitService.updateUnitStatusAndPublishEvents(UnitService.kt:391) at com.ingka.fms.inbound.reversereceiving.common.service.UnitService.updateUnitStatusAndPublishEvent(UnitService.kt:253) at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103) at java.base/java.lang.reflect.Method.invoke(Method.java:580) at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:351) at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:196) at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:765) at org.springframework.transaction.interceptor.TransactionInterceptor$1.proceedWithInvocation(TransactionInterceptor.java:123) at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:392) at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:119) at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:184) at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:765) at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:717) at com.ingka.fms.inbound.reversereceiving.common.service.UnitService$$SpringCGLIB$$0.updateUnitStatusAndPublishEvent(<generated>) at com.ingka.fms.inbound.reversereceiving.receiving.service.ReceivingService.completeReceivingUnit(ReceivingService.kt:563) at com.ingka.fms.inbound.reversereceiving.receiving.service.ReceivingService.completeReceivingUnits(ReceivingService.kt:554) at com.ingka.fms.inbound.reversereceiving.receiving.service.ReceivingService.completeReceiving$lambda$8(ReceivingService.kt:298) at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1768) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642) at java.base/java.lang.Thread.run(Thread.java:1583) Caused by: com.solace.spring.cloud.stream.binder.util.ClosedChannelBindingException: Message handler 028dfc28-458f-46d9-a669-aa22169323fd is not running ... 41 common frames omitted ``` maybe it gives more clue I do see this in the logs as well: ``` Stopping producer to TOPIC xxx/xxx/status/updated/V2/FA/FMSRR/xx/xx/ddd/aaaa/rc <message handler ID: 088fd415-8c5c-4ea0-91af-c319403403d3> ``` before the above log this log happens: ``` logger: c.s.s.c.s.b.util.SharedResourceManager message: 088fd415-8c5c-4ea0-91af-c319403403d3 is not the last user, persisting producer... ``` By a quick search in the Solace binder code, I see that this log comes from a stop() method, so I would assume that in a multithreaded environment, one thread would probably try to close the publisher while another thread is during its sending phase I also see this log: ``` logger: c.s.s.c.s.b.util.SharedResourceManager message: 088fd415-8c5c-4ea0-91af-c319403403d3 is not the last user, persisting producer... ``` ### Expected Behavior I expect Solace binder to publish the messages in a multi-threaded environment ### Steps to Reproduce 1. Create a thread pool 2. Try to send long messages using StreamBrdigeApi within the thread pool created in step 1 ### Solace Broker version _No response_ ### Solace API implementation("com.solace.spring.cloud:spring-cloud-starter-stream-solace:4.2.0") ### Solace API version implementation("com.solace.spring.cloud:spring-cloud-starter-stream-solace:4.2.0")

Topic		Replies	Views
Spring Cloud Stream exception: "doesn't support message NACK" after upgrading Solace Binder to v5.x Training, Tips and Tricks spring-cloud-stream , message-nack	1	54	October 21, 2024
Spring Cloud Stream Binder - Sending Persistent Messages with StreamBridge Connectors & Integrations spring , spring-cloud-stream	3	438	February 12, 2021
spring cloud stream solace pubsub PubSub+ Event Broker spring-cloud-stream , solacepubsub	5	83	August 29, 2024
Spring Cloud Stream, solace messages not going to error queue General Discussions	8	2452	February 17, 2022
Spring cloud bus General Discussions spring-cloud-stream	15	4589	February 7, 2022

Spring cloud stream solace binder StreamBridge.send() unexpected error

Related topics