JMS: Adding connection timeouts and retries (fixes #1192 and #1193) #1194

akara · 2018-09-06T19:40:48Z

All good for review now.

…at start and after failure.

akara · 2018-09-06T19:44:37Z

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

-    getAsyncCallback[JmsSession] { session =>
-      jmsSessions :+= session
-      onSessionOpened(session)
-    }


Moved to SourceStageLogic in JmsConsumerStage.scala for this function only applies to consumers

akara · 2018-09-06T19:45:15Z

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

-      case m: ActorMaterializer => m.system.dispatchers.lookup(dispatcher.dispatcher)
-      case x => throw new IllegalArgumentException(s"Stage only works with the ActorMaterializer, was: $x")
-    }
+    ActorMaterializerHelper.downcast(materializer).system.dispatchers.lookup(dispatcher.dispatcher)


Replaced with Akka library call providing the same functionality.

akara · 2018-09-06T21:20:49Z

Hmm, the build failure does not give me any indication what I could have messed up.

ennru · 2018-09-07T07:51:59Z

No worries, it's not related to your changes.
Please be aware of #1181 which will reach master before this.

ennru · 2018-09-07T07:54:21Z

fixes #1192 and #1193

ennru · 2018-09-07T07:55:31Z

fixes #1192 and #1193

akara · 2018-09-07T18:56:26Z

Thanks for the heads up! No worries, I'll rebase when time comes.

…connection exception issues using mock connections and artifacts

akara · 2018-09-09T19:15:51Z

Added tests with mock connection failure and start issues. Note this change only affects consumers for now. A bit more work required to make it work with producers.

ennru · 2018-09-10T06:33:35Z

The other JMS PR is now merged, you've compilation errors with Scala 2.11.

akara · 2018-09-10T17:48:52Z

We have been focusing on the consumer (Source) side more than the publishing side as a source fail will indeed lead to stream collapse, something we cannot have in a production environment. No decider can help with that. So the reconnect feature currently only applies to the consumer, while the connect timeout and retries apply to both.

Making the publisher safely deal with reconnects (without message losses) is not quite trivial, especially with the sessions pooling and publishes in-flight. I am contemplating whether publisher reconnects should be a separate task/PR altogether. Please voice your opinion.

ennru · 2018-09-11T14:49:59Z

Sounds good. Please, focus on the consumer side and leave the producer changes out.

# Conflicts: # jms/src/main/scala/akka/stream/alpakka/jms/JmsConnector.scala # jms/src/main/scala/akka/stream/alpakka/jms/JmsProducerStage.scala # jms/src/test/java/akka/stream/alpakka/jms/javadsl/JmsConnectorsTest.java

akara · 2018-09-11T23:58:54Z

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

@@ -552,12 +558,19 @@ class JmsConnectorsSpec extends JmsSpec with MockitoSugar {
        JmsProducerSettings(connectionFactory).withQueue("numbers")
      )

-      val completionFuture: Future[Done] = Source
-        .failed[JmsTextMessage](new RuntimeException("Simulated error"))


This test was unreliable. Since setting the jmsConnection in JmsConnector is done through an async callback, the early arrival of the failure will cause the jmsConnection (which is not referenced) to not be closed. Need to cause the failure mid-stream instead of at the beginning for this test to be reliable.

akara · 2018-09-12T00:04:12Z

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

+  case object Connected extends ConnectionStatus
+  case object TimedOut extends ConnectionStatus
+
+  protected def initSessionAsync(withReconnect: Boolean = true): Unit = {


withReconnect is set to false for the publisher at this point, until we can be certain reconnects behaving correctly with the publisher. Then even the if block below can be taken out.

# Conflicts: # jms/src/test/scala/akka/stream/alpakka/jms/scaladsl/JmsConnectorsSpec.scala

…e timing issues. The connection is ONLY used for shutdown. So no performance impact otherwise.

akara · 2018-09-13T05:05:11Z

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

@@ -30,10 +35,6 @@ private[jms] trait JmsConnector[S <: JmsSession] { this: GraphStageLogic =>

  protected val fail: AsyncCallback[Throwable] = getAsyncCallback[Throwable](e => failStage(e))

-  private val onConnection: AsyncCallback[jms.Connection] = getAsyncCallback[jms.Connection] { c =>
-    jmsConnection = Some(c)
-  }


Removed the connection setting using AsyncCallback and set directly instead. Some tests rely on the connection to be immediately available after establishing and fail if not. We minimize the risk of such failure. Also connection is only accessed at startup and shutdown. Making it @volatile does not incur runtime overhead.

Hmm, apparently this is still not enough. That test is still failing despite tightening up the timeline.

OK. Need to guarantee the connection is closed in an async system. To do so, we need to hold the connection in a future to ensure the connection will be closed even if it is resolved later than the postStop is called. This should settle it. But this test has creeped the scope of change.

ennru

Great work!
Yeah, these tests are hard to get super-stable.

ennru · 2018-09-13T07:21:20Z

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

+      if (status.compareAndSet(Connecting, TimedOut)) {
+        connectionRef.get.foreach(_.close())
+        connectionRef.set(None)
+        Future.failed(new TimeoutException("Timed out trying to establish connection"))


You could possibly add the applicable values in the error messages.

ennru

LGTM.
@andreas-schroeder Would you like to review, as well?

andreas-schroeder · 2018-09-15T15:32:40Z

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

+  def maxWaitTime: Duration =
+    if (maxRetries < 0) Duration.Inf
+    else {
+      val totalWaitTime = (0 to maxRetries).map(connectTimeout + waitTime(_)).reduce(_ + _)


super-cool idea to have that maxWaitTime on the settings to see how long it will wait in total!
How about
val totalWaitTime = (0 to maxRetries).map(connectTimeout + waitTime(_).min(maxBackoff)).reduce(_ + _)
to account for maxBackoff?

After thinking through this, we do not need the maxWaitTime any longer. It was used for establishing the blocking time waiting the connection in JmsProducerStage. After rebasing from your change where the connection is handled asynchronously there, there is no longer a need for this method. Let me remove it.

andreas-schroeder · 2018-09-15T15:36:13Z

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

+
+}
+
+case class ConnectionRetryException(message: String, cause: Throwable) extends Exception(message, cause)


how about moving that exception to JmsExceptions.scala? I think either all exceptions should live there, or all live in Jms.scala

Sure. Will do.

andreas-schroeder · 2018-09-15T16:03:02Z

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

+          else Future.failed(ConnectionRetryException(s"Could not establish connection after $n retries.", t))
+        } else {
+          val delay = if (maxed) maxBackoff else waitTime(nextN)
+          if (delay >= maxBackoff) {


why not just

val delay = waitTime(nextN).min(maxBackoff) after(delay, system.scheduler) { openConnectionWithRetry(startConnection, nextN) }

? :)

Since waitTime is an exponential function, there is a very small chance a very large backoff may overflow, throwing this logic into haywire. That's why I really want to stop calculating the waitTime after max is reached.

andreas-schroeder · 2018-09-15T16:03:55Z

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

@@ -208,7 +208,7 @@ final class JmsTxSourceStage(settings: JmsConsumerSettings)
 abstract class SourceStageLogic[T](shape: SourceShape[T],
                                   out: Outlet[T],
                                   settings: JmsConsumerSettings,
-                                   attributes: Attributes)
+                                   inheritedAttributes: Attributes)


great rename, that one was bothering me as well 👍

andreas-schroeder · 2018-09-15T16:08:53Z

jms/src/test/java/akka/stream/alpakka/jms/javadsl/JmsConnectorsTest.java

+          Throwable exception = tryResult.failed().get();
+          assertTrue(
+              "Did not fail with a ConnectionRetryException",
+              ConnectionRetryException.class.isAssignableFrom(exception.getClass()));


ain't that just exception instanceof ConnectionRetryException?

No, you're absolutely right. Was looking for a better way to express as well. Tells me my Java is getting a little rusty :-) Let me make this change, and the next one.

andreas-schroeder · 2018-09-15T16:09:33Z

jms/src/test/java/akka/stream/alpakka/jms/javadsl/JmsConnectorsTest.java

+              ConnectionRetryException.class.isAssignableFrom(exception.getClass()));
+          assertTrue(
+              "Cause of failure is not a JMSException",
+              JMSException.class.isAssignableFrom(exception.getCause().getClass()));


exception.getCause() instanceof JMSException, or am I wrong?

andreas-schroeder · 2018-09-15T16:10:40Z

jms/src/test/java/akka/stream/alpakka/jms/javadsl/JmsConnectorsTest.java

+              JMSException.class.isAssignableFrom(exception.getCause().getClass()));
+        });
+  }
+


👍 for that test. Great idea to check for the total time passed.

andreas-schroeder

I've commented on some (really) minor things I found while reading through the PR that could be either fixed right now, or in a subsequent change on the fly.
I would love to either build on top of that or help out on implementing the producer send retries.
Great work.

akara · 2018-09-15T17:59:05Z

Thank you very much for the comments @andreas-schroeder! Let me get the changes in by this weekend, before @ennru merges. Yes, we have to look at how to invalidate the in-flight publishes in case of a reconnect. Your expertise is greatly appreciated.

…on failed (connection dropped after creation), plugging the time gap of potential session failure after connection established and is closed.

ennru · 2018-09-17T06:57:59Z

Thank you for investing your efforts into the JMS connector!

JMS: Introducing connection establish timeout and connection retries …

46e4365

…at start and after failure.

akara commented Sep 6, 2018

View reviewed changes

ennru added the p:jms label Sep 7, 2018

Provided complete tests for hanging connection establish, start, and …

9af9e90

…connection exception issues using mock connections and artifacts

Updated tests to use syntax backwards-compatible with Scala 2.11.

80a7935

akara commented Sep 11, 2018

View reviewed changes

Removed unused traits.

8378e2c

akara commented Sep 12, 2018

View reviewed changes

Akara Sucharitakul added 5 commits September 12, 2018 09:57

Merge missed one test. Re-adding.

a4965e7

Merge remote-tracking branch 'upstream/master' into connection_retries

9e9b257

# Conflicts: # jms/src/test/scala/akka/stream/alpakka/jms/scaladsl/JmsConnectorsSpec.scala

Avoid duplicate tests.

7dc21a4

Provided documentation.

661f3da

Set jmsConnection directly after connection initialization to minimiz…

8df5f97

…e timing issues. The connection is ONLY used for shutdown. So no performance impact otherwise.

akara commented Sep 13, 2018

View reviewed changes

ennru reviewed Sep 13, 2018

View reviewed changes

Akara Sucharitakul added 3 commits September 13, 2018 10:01

Fixed failing test checking for closed connection.

5863959

Addressing review comments on timeout message.

4b45ce4

Auto-format the changed code to meet standards.

3e61d62

ennru approved these changes Sep 14, 2018

View reviewed changes

andreas-schroeder reviewed Sep 15, 2018

View reviewed changes

andreas-schroeder approved these changes Sep 15, 2018

View reviewed changes

Akara Sucharitakul added 2 commits September 16, 2018 10:34

Fixing as per PR comments by @andreas-schroeder.

7652c94

Fixing failed test. Also make sure to retry connect if session creati…

824b234

…on failed (connection dropped after creation), plugging the time gap of potential session failure after connection established and is closed.

ennru merged commit b8bef22 into akka:master Sep 17, 2018

ennru added this to the 0.21 milestone Sep 17, 2018

ennru mentioned this pull request Sep 17, 2018

JMS: Retry create connection on startup and connection failure #1193

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

JMS: Adding connection timeouts and retries (fixes #1192 and #1193) #1194

JMS: Adding connection timeouts and retries (fixes #1192 and #1193) #1194

akara commented Sep 6, 2018 •

edited

Loading

akara Sep 6, 2018

akara Sep 6, 2018

andreas-schroeder Sep 15, 2018

akara commented Sep 6, 2018

ennru commented Sep 7, 2018

ennru commented Sep 7, 2018

ennru commented Sep 7, 2018

akara commented Sep 7, 2018

akara commented Sep 9, 2018

ennru commented Sep 10, 2018

akara commented Sep 10, 2018

ennru commented Sep 11, 2018

akara Sep 11, 2018

akara Sep 12, 2018

akara Sep 13, 2018

akara Sep 13, 2018

akara Sep 13, 2018

ennru left a comment

ennru Sep 13, 2018

ennru left a comment

andreas-schroeder Sep 15, 2018

akara Sep 15, 2018

andreas-schroeder Sep 15, 2018

akara Sep 15, 2018

andreas-schroeder Sep 15, 2018

akara Sep 15, 2018

andreas-schroeder Sep 15, 2018

andreas-schroeder Sep 15, 2018

akara Sep 15, 2018

andreas-schroeder Sep 15, 2018

andreas-schroeder Sep 15, 2018

andreas-schroeder left a comment

akara commented Sep 15, 2018

ennru commented Sep 17, 2018


		}

		case class ConnectionRetryException(message: String, cause: Throwable) extends Exception(message, cause)

JMS: Adding connection timeouts and retries (fixes #1192 and #1193) #1194

JMS: Adding connection timeouts and retries (fixes #1192 and #1193) #1194

Conversation

akara commented Sep 6, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

akara commented Sep 6, 2018

ennru commented Sep 7, 2018

ennru commented Sep 7, 2018

ennru commented Sep 7, 2018

akara commented Sep 7, 2018

akara commented Sep 9, 2018

ennru commented Sep 10, 2018

akara commented Sep 10, 2018

ennru commented Sep 11, 2018

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

ennru left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

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

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 left a comment

Choose a reason for hiding this comment

akara commented Sep 15, 2018

ennru commented Sep 17, 2018

akara commented Sep 6, 2018 •

edited

Loading