inconsistent RejectedExecutionException behavior with Future.sequence

[achyan]

Reproduction steps

Scala version: 2.13.12

import java.util.concurrent._
import scala.concurrent.{Await, ExecutionContext, Future}

val threadPoolExecutor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS, new ArrayBlockingQueue(1))
val threadPool:ExecutionContext =  ExecutionContext.fromExecutor(threadPoolExecutor,
    t => { throw(t) })

val future1 = Future({ Thread.sleep(9999999)})(threadPool)
val future2 = Future({ Thread.sleep(9999999)})(threadPool)
val future3 = Future({ Thread.sleep(9999999)})(threadPool) // Future(Failure) due to RejectedRequest
implicit val implicitThreadpool: ExecutionContext = threadPool

Future.sequence(Seq(future1)) // Future(<not completed>) -- fine
Future.sequence(Seq(future3)) // Future(Failure(java.util.concurrent.RejectedExecutionException: Task Future(<not completed>) rejected from java.util.concurrent.ThreadPoolExecutor@416a4275[Running, pool size = 1, active threads = 1, queued tasks = 1, completed tasks = 0])) -- fine

Future.sequence(Seq(future3, future1)) // val res2: scala.concurrent.Future[Seq[Unit]] = Future(Failure(java.util.concurrent.RejectedExecutionException: Task Future(<not completed>) rejected from java.util.concurrent.ThreadPoolExecutor@416a4275[Running, pool size = 1, active threads = 1, queued tasks = 1, completed tasks = 0])) -- fine

Future.sequence(Seq(future1, future3)) 
/*
java.util.concurrent.RejectedExecutionException: Task Future(<not completed>) rejected from java.util.concurrent.ThreadPoolExecutor@416a4275[Running, pool size = 1, active threads = 1, queued tasks = 1, completed tasks = 0]
  at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
  at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
  at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
  at scala.concurrent.impl.ExecutionContextImpl.execute(ExecutionContextImpl.scala:21)
  at scala.concurrent.impl.Promise$Transformation.submitWithValue(Promise.scala:429)
  at scala.concurrent.impl.Promise$DefaultPromise.submitWithValue(Promise.scala:338)
  at scala.concurrent.impl.Promise$DefaultPromise.dispatchOrAddCallbacks(Promise.scala:312)
  at scala.concurrent.impl.Promise$DefaultPromise.onComplete(Promise.scala:216)
  at scala.concurrent.impl.Promise$DefaultPromise.zipWith(Promise.scala:164)
  at scala.concurrent.Future$.$anonfun$sequence$1(Future.scala:720)
  at scala.collection.IterableOnceOps.foldLeft(IterableOnce.scala:676)
  at scala.collection.IterableOnceOps.foldLeft$(IterableOnce.scala:670)
  at scala.collection.AbstractIterator.foldLeft(Iterator.scala:1300)
  at scala.concurrent.Future$.sequence(Future.scala:720)
  */

Problem

I would've expected Future.sequence(Seq(future3, future1)) to match Future.sequence(Seq(future1, future3)) in either both returning an exception or both returning a failed future.

Also, with Scala 2.12, we would get the RejectedExecutionException when we submitted the 3rd future that exceeded the thread pool's queue limit. With Scala 2.13, we now get back a Future(Failure(ex)) instead of the exception. If that's the case, I would've expected the Future.sequence to also not throw an exception.

[SethTisue]

We have a "scala/collections" team of volunteers we can summon on collections stuff. This ticket is a good example of why we ought to assemble a similar "scala/concurrency" team, as I bet there are people out there who might have some insight into this, but they apparently aren't watching scala/bug.

[He-Pin]

This can be configured with the default forkjoinpool in Java 9 iirc.

[som-snytt]

The behavior is because sequence must execute on the execution context. The weird failure is due to the poorly configured EC.

Here is a better look:

// make it easier to see when the error happens
val threadPool: ExecutionContext =  ExecutionContext.fromExecutor(threadPoolExecutor, t => println(s"ERROR ${t.getMessage}"))

// helper to pass in a specific EC
def seqtest[A, CC[X] <: IterableOnce[X], To](in: CC[Future[A]])(ec: ExecutionContext)(implicit bf: BuildFrom[CC[Future[A]], A, To]): Future[To] = Future.sequence(in)(bf, ec)

val fs = Try(seqtest(Seq(future1, future3))(global))
val gs = Try(seqtest(Seq(future3, future1))(global))
val hs = Try(seqtest(Seq(future1, future3))(threadPool))
val is = Try(seqtest(Seq(future3, future1))(threadPool))

You'll see ERROR Task Future(<not completed>) due to the failed sequence.

The sequence on global will print the expected failure twice:

Success(Future(Failure(java.util.concurrent.RejectedExecutionException: Task Future(<not completed>)

The bad sequence will print:

Success(Future(<not completed>))

for the first one, "1 then 3". The reason is that "3 then 1" is an error right away, but "1 then 3" has to submit an extra job to combine the results, and that job fails.

The other workaround is to use ExecutionContext.parasitic for the sequence.

The question is whether this is the same problem as #9071.

After all that verbiage, it seems to be a regression in scala/scala#9655. It works on 2.13.6 or stubbing Haoyi's override.