This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new 7100b50d6 [jms] close sessions when exceptions happen (#485)
7100b50d6 is described below
commit 7100b50d6d3d1850a4909859c540a7aee1cd8e3c
Author: PJ Fanning <[email protected]>
AuthorDate: Thu Feb 22 23:41:53 2024 +0100
[jms] close sessions when exceptions happen (#485)
* [jms] close sessions when exceptions happen
* add jms tests
* Update JmsProducerRetrySpec.scala
* update tests
* Update Dependencies.scala
* update tests
* Update JmsProducerRetrySpec.scala
---
.../stream/connectors/jms/impl/JmsConnector.scala | 2 +-
.../connectors/jms/JmsConnectionStatusSpec.scala | 8 +++++-
.../connectors/jms/JmsProducerRetrySpec.scala | 30 +++++++++++++++++++---
.../pekko/stream/connectors/jms/JmsSpec.scala | 1 +
.../jms/scaladsl/CachedConnectionFactory.scala | 6 ++---
project/Dependencies.scala | 3 ++-
6 files changed, 39 insertions(+), 11 deletions(-)
diff --git
a/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/JmsConnector.scala
b/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/JmsConnector.scala
index 5fa898724..3ddd31f0c 100644
---
a/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/JmsConnector.scala
+++
b/jms/src/main/scala/org/apache/pekko/stream/connectors/jms/impl/JmsConnector.scala
@@ -145,7 +145,7 @@ private[jms] trait JmsConnector[S <: JmsSession] {
}
private def handleRetriableException(ex: Throwable): Unit = {
- jmsSessions = Seq.empty
+ closeSessions()
connectionState match {
case JmsConnectorInitializing(_, attempt, backoffMaxed, _) =>
maybeReconnect(ex, attempt, backoffMaxed)
diff --git
a/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsConnectionStatusSpec.scala
b/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsConnectionStatusSpec.scala
index 64ed2d00d..a16cb6575 100644
---
a/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsConnectionStatusSpec.scala
+++
b/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsConnectionStatusSpec.scala
@@ -21,6 +21,8 @@ import pekko.stream.OverflowStrategy
import pekko.stream.connectors.jms.scaladsl.JmsConnectorState._
import pekko.stream.connectors.jms.scaladsl.{ JmsConnectorState, JmsConsumer,
JmsProducer, JmsProducerStatus }
import pekko.stream.scaladsl.{ Flow, Keep, Sink, SinkQueueWithCancel, Source }
+
+import com.github.pjfanning.jmswrapper.WrappedConnectionFactory
import javax.jms._
import org.mockito.ArgumentMatchers.{ any, anyBoolean, anyInt }
import org.mockito.Mockito._
@@ -38,8 +40,9 @@ class JmsConnectionStatusSpec extends JmsSpec {
"report disconnected on producer stream failure" in
withConnectionFactory() { connectionFactory =>
val connectedLatch = new CountDownLatch(1)
+ val wrappedConnectionFactory = new
WrappedConnectionFactory(connectionFactory)
- val jmsSink = textSink(JmsProducerSettings(producerConfig,
connectionFactory).withQueue("test"))
+ val jmsSink = textSink(JmsProducerSettings(producerConfig,
wrappedConnectionFactory).withQueue("test"))
val exception = new RuntimeException("failing stage")
val producerStatus = Source
@@ -63,6 +66,9 @@ class JmsConnectionStatusSpec extends JmsSpec {
status should havePublishedState(Failing(exception))
status should havePublishedState(Failed(exception))
+
+ wrappedConnectionFactory.getUnclosedSessionCount shouldBe 0
+ wrappedConnectionFactory.getUnclosedConnectionCount shouldBe 0
}
"report multiple connection attempts" in withMockedProducer { ctx =>
diff --git
a/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsProducerRetrySpec.scala
b/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsProducerRetrySpec.scala
index fa3c14fbe..1aa0fb1b8 100644
---
a/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsProducerRetrySpec.scala
+++
b/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsProducerRetrySpec.scala
@@ -19,6 +19,8 @@ import org.apache.pekko
import pekko.stream._
import pekko.stream.connectors.jms.scaladsl.{ JmsConsumer, JmsProducer }
import pekko.stream.scaladsl.{ Keep, Sink, Source }
+
+import com.github.pjfanning.jmswrapper.WrappedConnectionFactory
import javax.jms.{ JMSException, Message, TextMessage }
import org.mockito.ArgumentMatchers.{ any, anyInt, anyLong }
import org.mockito.Mockito.when
@@ -34,7 +36,7 @@ class JmsProducerRetrySpec extends JmsSpec {
"JmsProducer retries" should {
"retry sending on network failures" in withServer() { server =>
- val connectionFactory = server.createConnectionFactory
+ val connectionFactory = new
WrappedConnectionFactory(server.createConnectionFactory)
val jms = JmsProducer
.flow[JmsMapMessage](
JmsProducerSettings(producerConfig, connectionFactory)
@@ -90,10 +92,15 @@ class JmsProducerRetrySpec extends JmsSpec {
resultList.forall { produced =>
sentList.exists(consumed => index(consumed) == index(produced))
} shouldBe true
+
+ eventually {
+ connectionFactory.getUnclosedSessionCount shouldBe 0
+ connectionFactory.getUnclosedConnectionCount shouldBe 0
+ }
}
"fail sending only after max retries" in withServer() { server =>
- val connectionFactory = server.createConnectionFactory
+ val connectionFactory = new
WrappedConnectionFactory(server.createConnectionFactory)
val jms = JmsProducer
.flow[JmsMapMessage](
JmsProducerSettings(producerConfig, connectionFactory)
@@ -125,12 +132,16 @@ class JmsProducerRetrySpec extends JmsSpec {
val expectedDelay = 100L + 400L + 600L
failureTime - crashTime shouldBe >(expectedDelay)
failure shouldBe RetrySkippedOnMissingConnection
+
+ connectionFactory.getUnclosedSessionCount shouldBe 0
+ connectionFactory.getUnclosedConnectionCount shouldBe 0
}
"fail immediately on non-recoverable errors" in withConnectionFactory() {
connectionFactory =>
+ val wrappedConnectionFactory = new
WrappedConnectionFactory(connectionFactory)
val jms = JmsProducer
.flow[JmsMapMessage](
- JmsProducerSettings(producerConfig, connectionFactory)
+ JmsProducerSettings(producerConfig, wrappedConnectionFactory)
.withQueue("test")
.withSendRetrySettings(SendRetrySettings(system).withInfiniteRetries()))
.withAttributes(ActorAttributes.supervisionStrategy(stoppingDecider))
@@ -143,9 +154,15 @@ class JmsProducerRetrySpec extends JmsSpec {
val failure = result.failed.futureValue
failure shouldBe a[UnsupportedMapMessageEntryType]
+
+ eventually {
+ wrappedConnectionFactory.getUnclosedSessionCount shouldBe 0
+ wrappedConnectionFactory.getUnclosedConnectionCount shouldBe 0
+ }
}
"invoke supervisor when send fails" in withConnectionFactory() {
connectionFactory =>
+ val wrappedConnectionFactory = new
WrappedConnectionFactory(connectionFactory)
val deciderCalls = new AtomicInteger()
val decider: Supervision.Decider = { ex =>
deciderCalls.incrementAndGet()
@@ -154,7 +171,7 @@ class JmsProducerRetrySpec extends JmsSpec {
val jms = JmsProducer
.flow[JmsMapMessage](
- JmsProducerSettings(producerConfig, connectionFactory)
+ JmsProducerSettings(producerConfig, wrappedConnectionFactory)
.withQueue("test")
.withSendRetrySettings(SendRetrySettings(system).withInfiniteRetries()))
.withAttributes(ActorAttributes.supervisionStrategy(decider))
@@ -171,6 +188,11 @@ class JmsProducerRetrySpec extends JmsSpec {
list shouldBe List("1", "3")
deciderCalls.get shouldBe 1
+
+ eventually {
+ wrappedConnectionFactory.getUnclosedSessionCount shouldBe 0
+ wrappedConnectionFactory.getUnclosedConnectionCount shouldBe 0
+ }
}
"retry send as often as configured" in withMockedProducer { ctx =>
diff --git
a/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsSpec.scala
b/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsSpec.scala
index 24c698119..0ae12ec68 100644
--- a/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsSpec.scala
+++ b/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsSpec.scala
@@ -17,6 +17,7 @@ import org.apache.pekko
import pekko.actor.ActorSystem
import pekko.stream.connectors.testkit.scaladsl.LogCapturing
import pekko.testkit.TestKit
+
import javax.jms._
import jmstestkit.JmsBroker
import org.mockito.ArgumentMatchers.{ any, anyBoolean, anyInt }
diff --git
a/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/scaladsl/CachedConnectionFactory.scala
b/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/scaladsl/CachedConnectionFactory.scala
index 021f0dfc2..4a98489aa 100644
---
a/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/scaladsl/CachedConnectionFactory.scala
+++
b/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/scaladsl/CachedConnectionFactory.scala
@@ -15,18 +15,16 @@ package org.apache.pekko.stream.connectors.jms.scaladsl
import javax.jms.{ Connection, ConnectionFactory }
-import org.apache.activemq.ActiveMQConnection
-
/**
* a silly cached connection factory, not thread safe
*/
class CachedConnectionFactory(connFactory: ConnectionFactory) extends
ConnectionFactory {
- var cachedConnection: ActiveMQConnection = null
+ var cachedConnection: Connection = null
override def createConnection(): Connection = {
if (cachedConnection == null) {
- cachedConnection =
connFactory.createConnection().asInstanceOf[ActiveMQConnection]
+ cachedConnection = connFactory.createConnection()
}
cachedConnection
}
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index 81df7c03e..0675c1620 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -316,7 +316,8 @@ object Dependencies {
"com.ibm.mq" % "com.ibm.mq.allclient" % "9.3.4.1" % Test,
"org.apache.activemq" % "activemq-broker" % "5.16.7" % Test,
"org.apache.activemq" % "activemq-client" % "5.16.7" % Test,
- "io.github.sullis" %% "jms-testkit" % "1.0.4" % Test) ++ Mockito,
+ "io.github.sullis" %% "jms-testkit" % "1.0.4" % Test,
+ "com.github.pjfanning" % "jmswrapper" % "0.1.0" % Test) ++ Mockito,
// Having JBoss as a first resolver is a workaround for
https://github.com/coursier/coursier/issues/200
externalResolvers := ("jboss".at(
"https://repository.jboss.org/nexus/content/groups/public";)) +:
externalResolvers.value)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
