Flow has failed with error Shutting down because of violation of the Reactive Streams specification

Question

It seems I can never get the error handling right when using Akka Streams.

So this is my code

var db = Database.forConfig("oracle")
var mysqlDb = Database.forConfig("mysql_read")
var mysqlDbWrite = Database.forConfig("mysql_write")

implicit val actorSystem = ActorSystem()
val decider : Supervision.Decider = {
  case _: Exception =>
      println("got an exception restarting connections")
     // let us restart our connections
     db.close()
     mysqlDb.close()
     mysqlDbWrite.close()
     db = Database.forConfig("oracle")
     mysqlDb = Database.forConfig("mysql_read")
     mysqlDbWrite = Database.forConfig("mysql_write")
     Supervision.Restart
}
implicit val materializer = ActorMaterializer(ActorMaterializerSettings(actorSystem).withSupervisionStrategy(decider))

and I have a flow like this

val alreadyExistsFilter : Flow[Foo, Foo, NotUsed] = Flow[Foo].mapAsync(10){ foo =>
  try {
     val existsQuery = sql"""SELECT id FROM foo WHERE id = ${foo.id}""".as[Long]
     mysqlDbWrite.run(existsQuery).map(v => (foo, v))
  } catch {
     case e: Throwable =>
        println(s"Lookup failed for ${foo}")
        throw e // will restart the stream
  }
}.collect {case (f, v) if v.isEmpty => f}

So basically if the foo already exists in MySQL then the record should not be processed any further by the stream.

My hope with this code was that if anything fails with the mysql lookup (the mysql machine is pretty bad and timeouts are common), the record will be printed and discarded and the stream will continue with the remaining records courtesy of the supervision.

When I run this code. I see errors like

[error] (mysql_write network timeout executor) java.lang.RuntimeException: java.sql.SQLException: Invalid socket timeout value or state
java.lang.RuntimeException: java.sql.SQLException: Invalid socket timeout value or state
    at com.mysql.jdbc.ConnectionImpl$12.run(ConnectionImpl.java:5576)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.sql.SQLException: Invalid socket timeout value or state
    at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:998)
    at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:937)
    at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:926)
    at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:872)
    at com.mysql.jdbc.MysqlIO.setSocketTimeout(MysqlIO.java:4852)
    at com.mysql.jdbc.ConnectionImpl$12.run(ConnectionImpl.java:5574)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.net.SocketException: Socket is closed
    at java.net.Socket.setSoTimeout(Socket.java:1137)
    at com.mysql.jdbc.MysqlIO.setSocketTimeout(MysqlIO.java:4850)
    at com.mysql.jdbc.ConnectionImpl$12.run(ConnectionImpl.java:5574)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

and

[error] (mysql_write network timeout executor) java.lang.NullPointerException
java.lang.NullPointerException
    at com.mysql.jdbc.MysqlIO.setSocketTimeout(MysqlIO.java:4850)
    at com.mysql.jdbc.ConnectionImpl$12.run(ConnectionImpl.java:5574)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

One thing which surprises me here is that these exceptions don't come from my catch block. because I don't see the println statement of my catch block. The stack trace doesn't show me where it originated from... but since it is saying mysql_write I can assume that its the Flow above because only this Flow uses mysql_write.

Finally the entire stream crashes with the error

[trace] Stack trace suppressed: run last compile:runMain for the full output.
flow has failed with error Shutting down because of violation of the Reactive Streams specification.
14:51:06,973 |-INFO in ch.qos.logback.classic.AsyncAppender[asyncKafkaAppender] - Worker thread will flush remaining events before exiting.
[success] Total time: 3480 s, completed Sep 26, 2017 2:51:07 PM
14:51:07,603 |-INFO in ch.qos.logback.core.hook.DelayingShutdownHook@2320545b - Sleeping for 1 seconds

I don't know what I did to violate the reactive streams specification!!

Answer 1

A first stab at getting a more predictable solution would be removing the blocking behaviour (Await.result) and use mapAsync. A rewrite of the alreadyExistsFilter flow could be:

  val alreadyExistsFilter : Flow[Foo, Foo, NotUsed] = Flow[Foo].mapAsync(3) { foo ⇒
    val existsQuery = sql"""SELECT id FROM foo WHERE id = ${foo.id}""".as[Long]
    foo → Await.result(mysqlDbWrite.run(existsQuery), Duration.Inf)
  }.collect{
    case (foo, res) if res.isDefined ⇒ foo
  }

More info on blocking in Akka can be found in the docs.