pjfanning commented on code in PR #2063:
URL: https://github.com/apache/pekko-connectors/pull/2063#discussion_r4499353825
##########
amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpConnectionProvidersSpec.scala:
##########
@@ -147,6 +151,46 @@ class AmqpConnectionProvidersSpec extends AmqpSpec {
connection1.isOpen should be(false)
connection2.isOpen should be(false)
}
+
+ "replace a closed connection" in {
+ val reusableConnectionProvider =
AmqpCachedConnectionProvider(AmqpLocalConnectionProvider)
+ val closedConnection = reusableConnectionProvider.get
+ closedConnection.close()
+ val connection = reusableConnectionProvider.get
+ (connection should not).equal(closedConnection)
+ reusableConnectionProvider.release(closedConnection)
+ connection.isOpen should be(true)
+ reusableConnectionProvider.release(connection)
+ connection.isOpen should be(false)
+ }
+
+ "block concurrent clients while the connection is being opened" in {
+ val connecting = new CountDownLatch(1)
+ val connect = new CountDownLatch(1)
+ val connectionFactory = new ConnectionFactory() {
+ override def newConnection(es: ExecutorService, ar: AddressResolver,
name: String): Connection = {
+ connecting.countDown()
+ connect.await()
+ super.newConnection(es, ar, name)
+ }
+ }
+ val reusableConnectionProvider = AmqpCachedConnectionProvider(
+
AmqpConnectionFactoryConnectionProvider(connectionFactory).withHostAndPort("localhost",
5672))
+ val connections = new ConcurrentLinkedQueue[Connection]()
+ val first = new Thread(() =>
connections.add(reusableConnectionProvider.get))
+ val second = new Thread(() =>
connections.add(reusableConnectionProvider.get))
Review Comment:
This doesn't compile on Scala 3, which is what's failing the `connectors
(amqp)` job: `[E134] None of the overloaded alternatives of constructor Thread
... match arguments (() => Boolean)`. `ConcurrentLinkedQueue.add` returns
`Boolean`, so the lambda needs to return `Unit`.
```suggestion
val first = new Thread(() => {
connections.add(reusableConnectionProvider.get); () })
val second = new Thread(() => {
connections.add(reusableConnectionProvider.get); () })
```
##########
amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/AmqpConnectionProvider.scala:
##########
@@ -368,62 +365,36 @@ final class AmqpCachedConnectionProvider private (val
provider: AmqpConnectionPr
val automaticRelease: Boolean = true)
extends AmqpConnectionProvider {
- import pekko.stream.connectors.amqp.AmqpCachedConnectionProvider._
- private val state = new AtomicReference[State](Empty)
+ private val lock = new Object
+ private var cached: Option[Connection] = None
+ private var clients = Map.empty[Connection, Int]
def withAutomaticRelease(automaticRelease: Boolean):
AmqpCachedConnectionProvider =
copy(automaticRelease = automaticRelease)
- override def get: Connection = getRecursive(provider)
-
- @tailrec
- private def getRecursive(amqpConnectionProvider: AmqpConnectionProvider):
Connection = state.get match {
- case Empty =>
- if (state.compareAndSet(Empty, Connecting)) {
- try {
- val connection = amqpConnectionProvider.get
- if (!state.compareAndSet(Connecting, Connected(connection, 1)))
- throw new ConcurrentModificationException(
- "Unexpected concurrent modification while creating the
connection.")
- connection
- } catch {
- case e: ConcurrentModificationException => throw e
- case e: Throwable =>
- state.compareAndSet(Connecting, Empty)
- throw e
- }
- } else getRecursive(amqpConnectionProvider)
- case Connecting =>
getRecursive(amqpConnectionProvider)
- case c @ Connected(connection, clients) =>
- if (state.compareAndSet(c, Connected(connection, clients + 1)))
connection
- else getRecursive(amqpConnectionProvider)
- case Closing => getRecursive(amqpConnectionProvider)
+ override def get: Connection = lock.synchronized {
+ val connection = cached.filter(_.isOpen).getOrElse {
+ val connection = provider.get
Review Comment:
This runs a blocking connect while holding the lock, and `provider.release`
on line 393 does the same with a blocking close. If the broker is unreachable,
one caller can hold the lock for the whole connection timeout (60s by default).
Meanwhile:
- Each `release`, even a plain count decrement, blocks a thread for that
time. In the old `Connected` state it never waited.
- Queued `get` calls retry the failing connect one at a time, so the last
caller could wait about N x timeout.
- Waiters can't be interrupted or time out, which can also delay shutdown.
Could the bookkeeping stay under the lock and the I/O run outside it, with
waiters sharing a single connection attempt (for example a
`CompletableFuture`), so that one failure fails them all? That would also avoid
holding a `synchronized` monitor across blocking I/O, which pins virtual-thread
carriers on JDK versions below 24.
##########
amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/AmqpConnectionProvider.scala:
##########
@@ -368,62 +365,36 @@ final class AmqpCachedConnectionProvider private (val
provider: AmqpConnectionPr
val automaticRelease: Boolean = true)
extends AmqpConnectionProvider {
- import pekko.stream.connectors.amqp.AmqpCachedConnectionProvider._
- private val state = new AtomicReference[State](Empty)
+ private val lock = new Object
+ private var cached: Option[Connection] = None
+ private var clients = Map.empty[Connection, Int]
def withAutomaticRelease(automaticRelease: Boolean):
AmqpCachedConnectionProvider =
copy(automaticRelease = automaticRelease)
- override def get: Connection = getRecursive(provider)
-
- @tailrec
- private def getRecursive(amqpConnectionProvider: AmqpConnectionProvider):
Connection = state.get match {
- case Empty =>
- if (state.compareAndSet(Empty, Connecting)) {
- try {
- val connection = amqpConnectionProvider.get
- if (!state.compareAndSet(Connecting, Connected(connection, 1)))
- throw new ConcurrentModificationException(
- "Unexpected concurrent modification while creating the
connection.")
- connection
- } catch {
- case e: ConcurrentModificationException => throw e
- case e: Throwable =>
- state.compareAndSet(Connecting, Empty)
- throw e
- }
- } else getRecursive(amqpConnectionProvider)
- case Connecting =>
getRecursive(amqpConnectionProvider)
- case c @ Connected(connection, clients) =>
- if (state.compareAndSet(c, Connected(connection, clients + 1)))
connection
- else getRecursive(amqpConnectionProvider)
- case Closing => getRecursive(amqpConnectionProvider)
+ override def get: Connection = lock.synchronized {
+ val connection = cached.filter(_.isOpen).getOrElse {
Review Comment:
`isOpen` can't tell a permanently closed connection from one that's
recovering. `ConnectionFactory` enables automatic recovery by default, and
`AutorecoveringConnection.isOpen` just delegates to the current underlying
connection, so it should return false during a recovery attempt.
A `get` during a brief network drop would therefore open connection B and
overwrite `cached`, while existing stages stay on A, which then recovers. The
provider ends up with two live broker connections despite promising one shared
connection, and connection-scoped resources such as exclusive queues are split
between them.
Could this distinguish the two cases, for example by checking
`com.rabbitmq.client.Recoverable` or tracking recovery with a
`RecoveryListener`? A test using a recovering connection would be useful too.
##########
amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/AmqpConnectionProvider.scala:
##########
@@ -368,62 +365,36 @@ final class AmqpCachedConnectionProvider private (val
provider: AmqpConnectionPr
val automaticRelease: Boolean = true)
extends AmqpConnectionProvider {
- import pekko.stream.connectors.amqp.AmqpCachedConnectionProvider._
- private val state = new AtomicReference[State](Empty)
+ private val lock = new Object
+ private var cached: Option[Connection] = None
+ private var clients = Map.empty[Connection, Int]
def withAutomaticRelease(automaticRelease: Boolean):
AmqpCachedConnectionProvider =
copy(automaticRelease = automaticRelease)
- override def get: Connection = getRecursive(provider)
-
- @tailrec
- private def getRecursive(amqpConnectionProvider: AmqpConnectionProvider):
Connection = state.get match {
- case Empty =>
- if (state.compareAndSet(Empty, Connecting)) {
- try {
- val connection = amqpConnectionProvider.get
- if (!state.compareAndSet(Connecting, Connected(connection, 1)))
- throw new ConcurrentModificationException(
- "Unexpected concurrent modification while creating the
connection.")
- connection
- } catch {
- case e: ConcurrentModificationException => throw e
- case e: Throwable =>
- state.compareAndSet(Connecting, Empty)
- throw e
- }
- } else getRecursive(amqpConnectionProvider)
- case Connecting =>
getRecursive(amqpConnectionProvider)
- case c @ Connected(connection, clients) =>
- if (state.compareAndSet(c, Connected(connection, clients + 1)))
connection
- else getRecursive(amqpConnectionProvider)
- case Closing => getRecursive(amqpConnectionProvider)
+ override def get: Connection = lock.synchronized {
+ val connection = cached.filter(_.isOpen).getOrElse {
+ val connection = provider.get
+ cached = Some(connection)
+ connection
+ }
+ clients = clients.updated(connection, clients.getOrElse(connection, 0) + 1)
+ connection
}
- override def release(connection: Connection): Unit =
releaseRecursive(provider, connection)
-
- @tailrec
- private def releaseRecursive(amqpConnectionProvider: AmqpConnectionProvider,
connection: Connection): Unit =
- state.get match {
- case Empty => throw new
IllegalStateException("There is no connection to release.")
- case Connecting =>
releaseRecursive(amqpConnectionProvider, connection)
- case c @ Connected(cachedConnection, clients) =>
- if (cachedConnection != connection)
- throw new IllegalArgumentException("Can't release a connection
that's not owned by this provider")
-
- if (clients == 1 || !automaticRelease) {
- if (state.compareAndSet(c, Closing)) {
- amqpConnectionProvider.release(connection)
- if (!state.compareAndSet(Closing, Empty))
- throw new ConcurrentModificationException(
- "Unexpected concurrent modification while closing the
connection.")
- }
- } else {
- if (!state.compareAndSet(c, Connected(cachedConnection, clients -
1)))
- releaseRecursive(amqpConnectionProvider, connection)
- }
- case Closing => releaseRecursive(amqpConnectionProvider, connection)
+ override def release(connection: Connection): Unit = lock.synchronized {
+ if (clients.isEmpty) throw new IllegalStateException("There is no
connection to release.")
+ clients.get(connection) match {
+ case None =>
+ throw new IllegalArgumentException("Can't release a connection that's
not owned by this provider")
+ case Some(count) if count == 1 || !automaticRelease =>
Review Comment:
This changes behaviour for `automaticRelease = false` once a connection has
been replaced. If a user keeps `conn = provider.get` (A) and calls
`provider.release(conn)` at shutdown, that now removes only A. A replacement B
created by a restarted stage stays open, and the user has no handle to close it.
A second `release(A)` also now throws `IllegalArgumentException` rather than
the previous `IllegalStateException`.
Should releasing in this mode close every connection the provider tracks?
Either way, this case needs a test, as the new replacement test only covers
automatic release.
##########
amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpConnectionProvidersSpec.scala:
##########
@@ -147,6 +151,46 @@ class AmqpConnectionProvidersSpec extends AmqpSpec {
connection1.isOpen should be(false)
connection2.isOpen should be(false)
}
+
+ "replace a closed connection" in {
+ val reusableConnectionProvider =
AmqpCachedConnectionProvider(AmqpLocalConnectionProvider)
+ val closedConnection = reusableConnectionProvider.get
+ closedConnection.close()
+ val connection = reusableConnectionProvider.get
+ (connection should not).equal(closedConnection)
+ reusableConnectionProvider.release(closedConnection)
+ connection.isOpen should be(true)
+ reusableConnectionProvider.release(connection)
+ connection.isOpen should be(false)
+ }
+
+ "block concurrent clients while the connection is being opened" in {
+ val connecting = new CountDownLatch(1)
+ val connect = new CountDownLatch(1)
+ val connectionFactory = new ConnectionFactory() {
+ override def newConnection(es: ExecutorService, ar: AddressResolver,
name: String): Connection = {
+ connecting.countDown()
+ connect.await()
+ super.newConnection(es, ar, name)
+ }
+ }
+ val reusableConnectionProvider = AmqpCachedConnectionProvider(
+
AmqpConnectionFactoryConnectionProvider(connectionFactory).withHostAndPort("localhost",
5672))
+ val connections = new ConcurrentLinkedQueue[Connection]()
+ val first = new Thread(() =>
connections.add(reusableConnectionProvider.get))
+ val second = new Thread(() =>
connections.add(reusableConnectionProvider.get))
+ try {
+ first.start()
+ connecting.await()
Review Comment:
This wait has no timeout, and neither do `first.join()` and `second.join()`
below. If `first` fails before calling `newConnection`, this blocks until CI
times out.
Exceptions in the worker threads are also lost: with the broker down, `have
size 1` reports size 0 and the real cause never appears. Could this capture
thrown errors and use timed `await` and `join`?
##########
amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpConnectionProvidersSpec.scala:
##########
@@ -147,6 +151,46 @@ class AmqpConnectionProvidersSpec extends AmqpSpec {
connection1.isOpen should be(false)
connection2.isOpen should be(false)
}
+
+ "replace a closed connection" in {
+ val reusableConnectionProvider =
AmqpCachedConnectionProvider(AmqpLocalConnectionProvider)
+ val closedConnection = reusableConnectionProvider.get
+ closedConnection.close()
+ val connection = reusableConnectionProvider.get
+ (connection should not).equal(closedConnection)
+ reusableConnectionProvider.release(closedConnection)
+ connection.isOpen should be(true)
+ reusableConnectionProvider.release(connection)
+ connection.isOpen should be(false)
+ }
+
+ "block concurrent clients while the connection is being opened" in {
+ val connecting = new CountDownLatch(1)
+ val connect = new CountDownLatch(1)
+ val connectionFactory = new ConnectionFactory() {
+ override def newConnection(es: ExecutorService, ar: AddressResolver,
name: String): Connection = {
+ connecting.countDown()
+ connect.await()
+ super.newConnection(es, ar, name)
+ }
+ }
+ val reusableConnectionProvider = AmqpCachedConnectionProvider(
+
AmqpConnectionFactoryConnectionProvider(connectionFactory).withHostAndPort("localhost",
5672))
+ val connections = new ConcurrentLinkedQueue[Connection]()
+ val first = new Thread(() =>
connections.add(reusableConnectionProvider.get))
+ val second = new Thread(() =>
connections.add(reusableConnectionProvider.get))
+ try {
+ first.start()
+ connecting.await()
+ second.start()
+ eventually(second.getState should be(Thread.State.BLOCKED))
Review Comment:
This spec doesn't override `patienceConfig`, so `eventually` uses
ScalaTest's 150ms default, which could fail intermittently on a loaded runner.
The other AMQP specs set `PatienceConfig(10.seconds)`.
`BLOCKED` is also a weak signal, as it only shows the thread is waiting on
some monitor, such as during class loading, and not necessarily this provider's
lock. Asserting that the factory's connect count stays at 1 while `second`
waits would test the intended behaviour more directly.
##########
amqp/src/main/scala/org/apache/pekko/stream/connectors/amqp/AmqpConnectionProvider.scala:
##########
@@ -368,62 +365,36 @@ final class AmqpCachedConnectionProvider private (val
provider: AmqpConnectionPr
val automaticRelease: Boolean = true)
extends AmqpConnectionProvider {
- import pekko.stream.connectors.amqp.AmqpCachedConnectionProvider._
- private val state = new AtomicReference[State](Empty)
+ private val lock = new Object
+ private var cached: Option[Connection] = None
+ private var clients = Map.empty[Connection, Int]
def withAutomaticRelease(automaticRelease: Boolean):
AmqpCachedConnectionProvider =
copy(automaticRelease = automaticRelease)
- override def get: Connection = getRecursive(provider)
-
- @tailrec
- private def getRecursive(amqpConnectionProvider: AmqpConnectionProvider):
Connection = state.get match {
- case Empty =>
- if (state.compareAndSet(Empty, Connecting)) {
- try {
- val connection = amqpConnectionProvider.get
- if (!state.compareAndSet(Connecting, Connected(connection, 1)))
- throw new ConcurrentModificationException(
- "Unexpected concurrent modification while creating the
connection.")
- connection
- } catch {
- case e: ConcurrentModificationException => throw e
- case e: Throwable =>
- state.compareAndSet(Connecting, Empty)
- throw e
- }
- } else getRecursive(amqpConnectionProvider)
- case Connecting =>
getRecursive(amqpConnectionProvider)
- case c @ Connected(connection, clients) =>
- if (state.compareAndSet(c, Connected(connection, clients + 1)))
connection
- else getRecursive(amqpConnectionProvider)
- case Closing => getRecursive(amqpConnectionProvider)
+ override def get: Connection = lock.synchronized {
+ val connection = cached.filter(_.isOpen).getOrElse {
+ val connection = provider.get
+ cached = Some(connection)
+ connection
+ }
+ clients = clients.updated(connection, clients.getOrElse(connection, 0) + 1)
+ connection
}
- override def release(connection: Connection): Unit =
releaseRecursive(provider, connection)
-
- @tailrec
- private def releaseRecursive(amqpConnectionProvider: AmqpConnectionProvider,
connection: Connection): Unit =
- state.get match {
- case Empty => throw new
IllegalStateException("There is no connection to release.")
- case Connecting =>
releaseRecursive(amqpConnectionProvider, connection)
- case c @ Connected(cachedConnection, clients) =>
- if (cachedConnection != connection)
- throw new IllegalArgumentException("Can't release a connection
that's not owned by this provider")
-
- if (clients == 1 || !automaticRelease) {
- if (state.compareAndSet(c, Closing)) {
- amqpConnectionProvider.release(connection)
- if (!state.compareAndSet(Closing, Empty))
- throw new ConcurrentModificationException(
- "Unexpected concurrent modification while closing the
connection.")
- }
- } else {
- if (!state.compareAndSet(c, Connected(cachedConnection, clients -
1)))
- releaseRecursive(amqpConnectionProvider, connection)
- }
- case Closing => releaseRecursive(amqpConnectionProvider, connection)
+ override def release(connection: Connection): Unit = lock.synchronized {
+ if (clients.isEmpty) throw new IllegalStateException("There is no
connection to release.")
+ clients.get(connection) match {
+ case None =>
+ throw new IllegalArgumentException("Can't release a connection that's
not owned by this provider")
+ case Some(count) if count == 1 || !automaticRelease =>
+ clients -= connection
+ if (cached.contains(connection)) cached = None
+ provider.release(connection)
Review Comment:
Following on from the `isOpen` comment: if A has been replaced and its last
client releases it during recovery, `clients -= connection` stops tracking it,
and the base `release` skips `close()` because `isOpen` is false. If recovery
then succeeds, A is open on the broker with no owner, so nothing can close it
for the life of the JVM.
This would need an unconditional close, or a recovery-aware check, for
connections that are being discarded.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
