Question

Send a response to Origin other than REST

  • 14 February 2023
  • 8 replies
  • 34 views
M
Rank

Hi

I need to connect to a WebSocket which will send me hundreds of messages in a TEXT format. Those messages need to be acknowledged so the web socket stops sending the same messages overtime. To acknowledge those messages, I need to send an identifier back to the Origin service in a self contained WebSocket text message.

So far, my pipeline is as per below:

 

However, I realized the “Send response to Origin” was not sending back the identifier to the WS.

After reading the documentation, it seems that I could only use this block in a microservice pipeline. But the microservice pipeline works on origins which listen/read JSON-formatted data. And that’s where I am blocked. Indeed, my messages from the Web sockets are NOT in JSON and it s not a REST service.

Is there a way to send a text message back to a Web socket in Streamset? So I can properly ACK each message that have been processed successfully.

Thank you 

8 replies

S
Rank
Userlevel 1
Badge

You can use the WebSocket destination to send data back. You will just need to configure it the same way as your origin. https://docs.streamsets.com/portal/platform-datacollector/latest/datacollector/UserGuide/Destinations/WebSocketClient.html

M
Rank

@sam I have seen this destination but there are 2 issues:

  • there is nowhere in the configuration of the destination to send a text message
  • is not this going to open a new web socket instead of reusing the same one
S
Rank
Userlevel 1
Badge

You can use the Data Format of “Text” to send a text message.

It will open a second WebSocket connection, it doesn’t reuse the connection.

M
Rank

The problem is that i m supposed to use the same web socket connection:

The WebSocket established with the URL is a textual bi-directional connection using UTF-8 encoding. The WebSocket service sends a sequence of notifications to the consumer and the consumer should send back a short acknowledgement over this connection for each notification received. This acknowledgement is for a particular notification and the server will periodically resend a notification until it is acknowledged. This will cease once the server has received and processed an acknowledgement for a particular notification.

 

Is this at all possible with Streamset DC?

S
Rank
Userlevel 1
Badge

If your server requires that the acknowledgement be sent using the same connection, then this cannot be done with Data Collector. Please test that it must use the same connection, many systems will still work if the acknowledgement is sent with a separate connection.

S
Rank
Userlevel 1
Badge

For anyone in the future, note that the WebSocket server can send responses back to the origin, but it always uses XML or JSON. https://docs.streamsets.com/platform-datacollector/latest/datacollector/UserGuide/Origins/WebSocketServer.html

M
Rank

I will check with the team if necessary for same connection or not.

In the meantime, the destination is giving me the below error for each messages, I dont understand what the error is?

 

Error Stack Trace

HTTP_50 - Error sending resource. Reason: java.io.IOException: /opt/streamsets-datacollector-5.3.0 is a directory
com.streamsets.pipeline.api.StageException: HTTP_50 - Error sending resource. Reason: java.io.IOException: /opt/streamsets-datacollector-5.3.0 is a directory	at com.streamsets.pipeline.stage.destination.websocket.WebSocketTarget.throwStageException(WebSocketTarget.java:202)	at com.streamsets.pipeline.stage.destination.websocket.WebSocketTarget.write(WebSocketTarget.java:180)	at com.streamsets.pipeline.api.base.configurablestage.DTarget.write(DTarget.java:34)	at com.streamsets.datacollector.runner.StageRuntime.lambda$execute$2(StageRuntime.java:291)	at com.streamsets.pipeline.api.impl.CreateByRef.call(CreateByRef.java:40)	at com.streamsets.datacollector.runner.StageRuntime.execute(StageRuntime.java:232)	at com.streamsets.datacollector.runner.StageRuntime.execute(StageRuntime.java:299)	at com.streamsets.datacollector.runner.StagePipe.process(StagePipe.java:209)	at com.streamsets.datacollector.execution.runner.common.ProductionPipelineRunner.processPipe(ProductionPipelineRunner.java:874)	at com.streamsets.datacollector.execution.runner.common.ProductionPipelineRunner.lambda$executeRunner$3(ProductionPipelineRunner.java:922)	at com.streamsets.datacollector.runner.PipeRunner.acceptConsumer(PipeRunner.java:195)	at com.streamsets.datacollector.runner.PipeRunner.forEachInternal(PipeRunner.java:140)	at com.streamsets.datacollector.runner.PipeRunner.executeBatch(PipeRunner.java:120)	at com.streamsets.datacollector.execution.runner.common.ProductionPipelineRunner.executeRunner(ProductionPipelineRunner.java:921)	at com.streamsets.datacollector.execution.runner.common.ProductionPipelineRunner.runSourceLessBatch(ProductionPipelineRunner.java:896)	at com.streamsets.datacollector.execution.runner.common.ProductionPipelineRunner.processBatch(ProductionPipelineRunner.java:511)	at com.streamsets.datacollector.runner.StageRuntime$3.run(StageRuntime.java:371)	at java.security.AccessController.doPrivileged(Native Method)	at com.streamsets.datacollector.runner.StageRuntime.processBatch(StageRuntime.java:367)	at com.streamsets.datacollector.runner.StageContext.processBatch(StageContext.java:334)	at com.streamsets.pipeline.stage.origin.websocket.WebSocketClientSource.process(WebSocketClientSource.java:273)	at com.streamsets.pipeline.stage.origin.websocket.WebSocketClientSource.onMessage(WebSocketClientSource.java:244)	at sun.reflect.GeneratedMethodAccessor495.invoke(Unknown Source)	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)	at java.lang.reflect.Method.invoke(Method.java:498)	at org.eclipse.jetty.websocket.common.events.annotated.CallableMethod.call(CallableMethod.java:71)	at org.eclipse.jetty.websocket.common.events.annotated.OptionalSessionCallableMethod.call(OptionalSessionCallableMethod.java:72)	at org.eclipse.jetty.websocket.common.events.JettyAnnotatedEventDriver.onTextMessage(JettyAnnotatedEventDriver.java:234)	at org.eclipse.jetty.websocket.common.message.SimpleTextMessage.messageComplete(SimpleTextMessage.java:69)	at org.eclipse.jetty.websocket.common.events.AbstractEventDriver.appendMessage(AbstractEventDriver.java:66)	at org.eclipse.jetty.websocket.common.events.JettyAnnotatedEventDriver.onTextFrame(JettyAnnotatedEventDriver.java:226)	at org.eclipse.jetty.websocket.common.events.AbstractEventDriver.incomingFrame(AbstractEventDriver.java:162)	at org.eclipse.jetty.websocket.common.WebSocketSession.incomingFrame(WebSocketSession.java:476)	at org.eclipse.jetty.websocket.common.extensions.ExtensionStack.incomingFrame(ExtensionStack.java:220)	at org.eclipse.jetty.websocket.common.Parser.notifyFrame(Parser.java:219)	at org.eclipse.jetty.websocket.common.Parser.parse(Parser.java:244)	at org.eclipse.jetty.websocket.common.io.AbstractWebSocketConnection.readParse(AbstractWebSocketConnection.java:560)	at org.eclipse.jetty.websocket.common.io.AbstractWebSocketConnection.onFillable(AbstractWebSocketConnection.java:391)	at org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:305)	at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:103)	at org.eclipse.jetty.io.ssl.SslConnection$DecryptedEndPoint.onFillable(SslConnection.java:411)	at org.eclipse.jetty.io.ssl.SslConnection.onFillable(SslConnection.java:305)	at org.eclipse.jetty.io.ssl.SslConnection$2.succeeded(SslConnection.java:159)	at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:103)	at org.eclipse.jetty.io.ChannelEndPoint$2.run(ChannelEndPoint.java:118)	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.runTask(EatWhatYouKill.java:333)	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.doProduce(EatWhatYouKill.java:310)	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.tryProduce(EatWhatYouKill.java:168)	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.run(EatWhatYouKill.java:126)	at org.eclipse.jetty.util.thread.ReservedThreadExecutor$ReservedThread.run(ReservedThreadExecutor.java:366)	at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:765)	at org.eclipse.jetty.util.thread.QueuedThreadPool$2.run(QueuedThreadPool.java:683)	at java.lang.Thread.run(Thread.java:750)Caused by: java.io.IOException: /opt/streamsets-datacollector-5.3.0 is a directory	at org.eclipse.jetty.util.resource.PathResource.getInputStream(PathResource.java:392)	at org.eclipse.jetty.util.security.CertificateUtils.getKeyStore(CertificateUtils.java:52)	at org.eclipse.jetty.util.ssl.SslContextFactory.loadTrustStore(SslContextFactory.java:1110)	at org.eclipse.jetty.util.ssl.SslContextFactory.load(SslContextFactory.java:276)	at org.eclipse.jetty.util.ssl.SslContextFactory.doStart(SslContextFactory.java:241)	at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68)	at org.eclipse.jetty.util.component.ContainerLifeCycle.start(ContainerLifeCycle.java:138)	at org.eclipse.jetty.util.component.ContainerLifeCycle.doStart(ContainerLifeCycle.java:108)	at org.eclipse.jetty.client.HttpClient.doStart(HttpClient.java:246)	at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68)	at org.eclipse.jetty.util.component.ContainerLifeCycle.start(ContainerLifeCycle.java:138)	at org.eclipse.jetty.util.component.ContainerLifeCycle.doStart(ContainerLifeCycle.java:108)	at org.eclipse.jetty.websocket.client.WebSocketClient.doStart(WebSocketClient.java:387)	at org.eclipse.jetty.util.component.AbstractLifeCycle.start(AbstractLifeCycle.java:68)	at com.streamsets.pipeline.stage.destination.websocket.WebSocketTarget.write(WebSocketTarget.java:165)	... 51 more
S
Rank
Userlevel 1
Badge

Please ask the error message question as a separate question and include the detail of the configuration. This will help the right people see it.

Reply


Terms & Conditions