JMS producer multi session #1148

andreas-schroeder · 2018-08-14T06:07:34Z

JmsProducerStage now supports multiple sessions by behaving
similarly to MapAsync
Producers are pooled and parallelism is controlled by how producers
are returned to the pool.
Created a separate JmsMessageProducer class to keep the stage logic
more focussed on coordinating elements within the stage.
Grouped the logic for creating the execution context to use in the
producer stage as well as in the consumer stage.

WellingR · 2018-08-15T18:05:32Z

jms/src/main/scala/akka/stream/alpakka/jms/JmsConnector.scala

+          case v: Byte => message.setByteProperty(key, v)
+          case v: Short => message.setShortProperty(key, v)
+          case v: Long => message.setLongProperty(key, v)
+          case v: Double => message.setDoubleProperty(key, v)


If the properties contain a value which is not of any of these types, would it make sense to explicitly throw an exception instead of the default MatchError?

In general, I would say yes.

However, the Jms producer crashes on any exception, and does not distinguish between retriable errors and non-retriable ones - it doesn't perform any retries. So in the end, the behavior would still be very similar if we introduced a custom exception here. It would make much more sense to introduce a custom exception if the Jms producer stage did retries, and would not try to do those retries if the property value type is of an unexpected type.

(please note: this is part of the code I moved out of the stage into a separate class - I didn't modify it)

@WellingR after giving it some thought, those exceptions can be handy in supervision logic too, so here is a proposal - what do you think? a280576

Looks better!
I had some comments about the exception traits though.

👍 I've followed your suggestions :)

ennru · 2018-08-20T17:56:07Z

Test compilation fails for Scala 2.11.

andreas-schroeder · 2018-08-21T05:52:55Z

@ennru thanks for notifying and sorry for not seeing this myself. I will take care of fixing the issues.

andreas-schroeder · 2018-08-22T05:14:32Z

@ennru compilation is now fixed, I just happen to add exception classes as suggested by @WellingR .

ennru

Great work!
I think your solution with the Holder and its callbacks deserves a description in comments so that others can understand it more easily.

ennru · 2018-08-24T07:13:59Z

jms/src/main/scala/akka/stream/alpakka/jms/JmsConnector.scala

@@ -40,11 +40,23 @@ private[jms] trait JmsConnector { this: GraphStageLogic =>
      onSessionOpened(session)
    }

-  private[jms] def initSessionAsync(dispatcher: Dispatcher): Future[Unit] = {
-    ec = materializer match {
+  private[jms] def executionContext(attributes: Attributes): ExecutionContext = {


This and a few other things in this class should be just regular private or protected.

agreed; I wanted to follow the style of the existing implementation, I'll change that to simpler visibility restrictions.

ennru · 2018-08-24T07:33:39Z

jms/src/main/scala/akka/stream/alpakka/jms/JmsProducerStage.scala

+
+      private val jmsProducers: Buffer[JmsMessageProducer] = Buffer(settings.sessionCount, settings.sessionCount)
+
+      private val buffer: Buffer[Holder[A]] = Buffer(settings.sessionCount, settings.sessionCount)


Would producerWithMessage describe better what it holds?

how about inFlightMessagesWithProducer?

ennru · 2018-08-24T07:42:28Z

jms/src/test/scala/akka/stream/alpakka/jms/scaladsl/JmsConnectorsSpec.scala

@@ -782,5 +785,92 @@ class JmsConnectorsSpec extends JmsSpec {

      result.futureValue should ===(input)
    }
+
+    "publish and consume strings through a queue with multiple sessions" in withServer() { ctx =>
+      //#connection-factory


You need to remove the copied //# as that marks what Paradox cuts out for the documentation.
It would be very useful if you added a section on the producer settings in the documentation.

I see, I wasn't sure what //# does. I'll add producer settings documentation as well.

WellingR · 2018-08-25T08:57:30Z

jms/src/main/scala/akka/stream/alpakka/jms/Exceptions.scala

+/**
+ * Marker trait indicating that the exception thrown is intermittent. The failed operation might succeed if tried again.
+ */
+trait RetriableException extends Exception


It makes some sense to mark some exception cases as NonRetriableException. However the other way around RetriableException does not make as much sense.

The essence is that you want to point out for some specific exceptions that a retry makes no sense. However for all other exceptions, a retry might make sense (so there is no real need for the marker trait RetriableException especially since the code does not even use this).

WellingR · 2018-08-25T09:00:16Z

jms/src/main/scala/akka/stream/alpakka/jms/Exceptions.scala

+/**
+ * Marker trait indicating that the exception thrown is persistent. The operation will always fail when retried.
+ */
+trait NonRetriableException extends Exception


Since this is a jms library, maybe we should call this a NonRetriableJmsException.

andreas-schroeder · 2018-08-25T13:00:15Z

Travis failed because an AMQP test failed: https://travis-ci.org/akka/alpakka/jobs/420470777#L680

andreas-schroeder · 2018-08-26T09:42:57Z

docs/src/main/paradox/jms.md

+  number of parallel sessions increases throughput at the cost of message ordering. While the messages may arrive
+  out of order at the Jms broker, the producer flow outputs messages in the order they are received.
+* `timeToLive`  (optional) the time messages should be kept on the Jms broker. This setting can be overridden on 
+  individual messages. If not set, messages will never expire.



Note that I didn't document acknowledgeMode. This is because I read various threads on Jms stating that this setting is not considered when producing records.

andreas-schroeder · 2018-08-27T13:47:58Z

Looks like it is always the same AMQP test that is consistently failing:
https://travis-ci.org/akka/alpakka/jobs/420947066#L631
https://travis-ci.org/akka/alpakka/jobs/420780309#L810
Should I rebase this PR on top of master?

ennru · 2018-08-27T14:02:48Z

Don't worry for the AMQP test, it is not your fault. See #1167.

ennru · 2018-08-28T10:22:07Z

The build failure is fixed in master, please re-base.

ennru

Just another tiny comment. LGTM, otherwise.

ennru · 2018-08-29T07:55:17Z

jms/src/main/scala/akka/stream/alpakka/jms/JmsConnector.scala

-      case settings: JmsConsumerSettings =>
-        settings.sessionCount
+      case settings: JmsConsumerSettings => settings.sessionCount
+      case settings: JmsProducerSettings => settings.sessionCount


You may push up sessionCount to JmsSettings to simplify this.

Will do :) I try to find some time today to finish things up.

Actually, the JmsBrowseSettings would then need to implement sessionCount as well, which would be hard-coded to 1. Should we go for that?

Yes, makes sense to me.

- created a separate JmsMessageProducer class to keep the stage logic more focussed on coordinating elements within the stage. - grouped the logic for creating the execution context to use in the producer stage as well as in the consumer stage.

- JmsProducerStage now supprots multiple sessions by behaving similarly to MapAsync - Producers are pooled and parallelism is controlled by how producers are returned to the pool. Fixes akka#1144

- Introduces traits to separate recoverable and non-recoverable errors - Introduces exceptions for populate map message and populate message properties - Refactors JmsMessageProducer to produce new exceptions

In order to explain the intent, added some brief comments to the marker traits.

- Adding tests for new exception types - Deleting retriable exception marker trait - renaming NonRetriableException to NonRetriableJmsException - Adding docs on producer settings - Adding comments to producer graph stage logic -

The number of sessions created did not reflect the desired sessionCount. Now it does.

The contributor advice states that case classes should not be used as settings objects. Therefore, this commit changes the producer config docs example from using the case class constructor to using the withXxx methods.

The previous implementation of the test was flaky, as it did assume that on failure of one send operation, the producer stage would emit all in-flight messages before the failure and then fail the stage. The actual behavior is fail-fast: failure of one send operation causes immediate failure of the stage. This commit adjusts the test to this behavior (that corresponds to MapAsync's behavior).

andreas-schroeder · 2018-08-29T20:11:31Z

Test failure is https://travis-ci.org/akka/alpakka/jobs/422226062#L715
I'm not 100% sure, but that test looks flaky to me, as the broker is stopped after the stream is started. If stopping the broker is delayed long enough for the stream to be actually started, a cached connection will be created. I can reproduce the failure with a Thread.sleep(1000) before ctx.broker.stop() here: https://github.com/akka/alpakka/blob/master/jms/src/test/scala/akka/stream/alpakka/jms/scaladsl/JmsConnectorsSpec.scala#L580

The tests "fail fast on the first failing send" and "sink disconnect exceptional completion" were racy. This commit fixes these tests by introducing explicit count down latches to wait for execution progress.

andreas-schroeder · 2018-08-30T05:21:25Z

I've changed the failing test and my test (that was actually racy too) in the hope of improving things.

andreas-schroeder · 2018-08-30T07:16:24Z

well, now I am at a loss at how to stabilize the failing test - any ideas welcome :)

ennru · 2018-08-30T08:48:08Z

We've seen failure on that from time to time: #1031
If you find a way to make it stable, please do (eg. in another PR).

andreas-schroeder · 2018-08-30T18:34:34Z

Now, it's an ftp test that fails: https://travis-ci.org/akka/alpakka/jobs/422635814#L644

ennru · 2018-08-31T07:37:57Z

Thank you for this great addition to the JMS connector!

- JmsProducerStage now supports multiple sessions by behaving similarly to MapAsync - Producers are pooled and parallelism is controlled by how producers are returned to the pool - Created a separate JmsMessageProducer class to keep the stage logic more focussed on coordinating elements within the stage. - Grouped the logic for creating the execution context to use in the producer stage as well as in the consumer stage. Fixes akka#1144

ennru added the p:jms label Aug 15, 2018

WellingR reviewed Aug 15, 2018

View reviewed changes

ennru reviewed Aug 24, 2018

View reviewed changes

WellingR reviewed Aug 25, 2018

View reviewed changes

andreas-schroeder commented Aug 26, 2018

View reviewed changes

ennru approved these changes Aug 29, 2018

View reviewed changes

andreas-sa-schroeder added 10 commits August 29, 2018 21:04

JMS allow using multiple sessions in producer

b2454cd

- JmsProducerStage now supprots multiple sessions by behaving similarly to MapAsync - Producers are pooled and parallelism is controlled by how producers are returned to the pool. Fixes akka#1144

Fix Scala 2.11 compilation issue

b1bef36

Add Jms Exception classes

34e040e

- Introduces traits to separate recoverable and non-recoverable errors - Introduces exceptions for populate map message and populate message properties - Refactors JmsMessageProducer to produce new exceptions

Add scaladocs to exception marker traits

e291278

In order to explain the intent, added some brief comments to the marker traits.

Adjust to review comments

e9e88b1

- Adding tests for new exception types - Deleting retriable exception marker trait - renaming NonRetriableException to NonRetriableJmsException - Adding docs on producer settings - Adding comments to producer graph stage logic -

Simplify visibility modifiers, add copyright notice

767c172

Jms: Honor producer settings sessionCount

5155e5d

The number of sessions created did not reflect the desired sessionCount. Now it does.

Avoid case class constructor in docs

54ffa1c

The contributor advice states that case classes should not be used as settings objects. Therefore, this commit changes the producer config docs example from using the case class constructor to using the withXxx methods.

andreas-schroeder force-pushed the jms-producer-multi-session branch from 799d93d to 17be020 Compare August 29, 2018 19:24

Reduce Jms test raciness

5d10d11

The tests "fail fast on the first failing send" and "sink disconnect exceptional completion" were racy. This commit fixes these tests by introducing explicit count down latches to wait for execution progress.

Jms: move sessionCount up to JmsSettings

2a990ff

ennru merged commit dccffeb into akka:master Aug 31, 2018

ennru added this to the 0.21 milestone Aug 31, 2018

andreas-schroeder deleted the jms-producer-multi-session branch October 17, 2018 08:50

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

JMS producer multi session #1148

JMS producer multi session #1148

andreas-schroeder commented Aug 14, 2018

WellingR Aug 15, 2018

andreas-schroeder Aug 20, 2018

andreas-schroeder Aug 22, 2018 •

edited

Loading

WellingR Aug 25, 2018

andreas-schroeder Aug 25, 2018

ennru commented Aug 20, 2018

andreas-schroeder commented Aug 21, 2018

andreas-schroeder commented Aug 22, 2018

ennru left a comment

ennru Aug 24, 2018

andreas-schroeder Aug 24, 2018

ennru Aug 24, 2018

andreas-schroeder Aug 24, 2018

ennru Aug 24, 2018

andreas-schroeder Aug 24, 2018

WellingR Aug 25, 2018 •

edited

Loading

WellingR Aug 25, 2018

andreas-schroeder commented Aug 25, 2018

andreas-schroeder Aug 26, 2018

andreas-schroeder commented Aug 27, 2018

ennru commented Aug 27, 2018

ennru commented Aug 28, 2018

ennru left a comment

ennru Aug 29, 2018

andreas-schroeder Aug 29, 2018

andreas-schroeder Aug 29, 2018

ennru Aug 30, 2018

andreas-schroeder commented Aug 29, 2018

andreas-schroeder commented Aug 30, 2018

andreas-schroeder commented Aug 30, 2018

ennru commented Aug 30, 2018

andreas-schroeder commented Aug 30, 2018

ennru commented Aug 31, 2018


		private val jmsProducers: Buffer[JmsMessageProducer] = Buffer(settings.sessionCount, settings.sessionCount)

		private val buffer: Buffer[Holder[A]] = Buffer(settings.sessionCount, settings.sessionCount)

JMS producer multi session #1148

JMS producer multi session #1148

Conversation

andreas-schroeder commented Aug 14, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andreas-schroeder Aug 22, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ennru commented Aug 20, 2018

andreas-schroeder commented Aug 21, 2018

andreas-schroeder commented Aug 22, 2018

ennru left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

WellingR Aug 25, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andreas-schroeder commented Aug 25, 2018

Choose a reason for hiding this comment

andreas-schroeder commented Aug 27, 2018

ennru commented Aug 27, 2018

ennru commented Aug 28, 2018

ennru left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

andreas-schroeder commented Aug 29, 2018

andreas-schroeder commented Aug 30, 2018

andreas-schroeder commented Aug 30, 2018

ennru commented Aug 30, 2018

andreas-schroeder commented Aug 30, 2018

ennru commented Aug 31, 2018

andreas-schroeder Aug 22, 2018 •

edited

Loading

WellingR Aug 25, 2018 •

edited

Loading