Unbinding a server will not close existing persistent connections

The below code reproduces a really strange bug I have seen with akka-http server. We have a number of unit tests which instantiate an HTTP server and make a request of it, testing for the expected response (which may have been either a failure or a success), shutting the server down after each test. I noticed that when I added a test which made a successful response, all future tests seem to "observe" the server from the successful test, even after it has been unbound.

Below I have a simple class called `AkkaBugRepr` which will bind an HTTP server when its `start` method is called. It can either return a piece of data or a 404 depending on how it is started. There is then a runnable test, `ReproduceAkkaBug`, which creates 5 instances of the server, makes a single request of each and tests that the response is as expected. The output should be 5 passed tests. Each test is run via

runSingleTest(fail = true|false)

If the method is called with `fail=true`, the test *should pass* (it is merely testing that the HTTP returns a 404). If the test is called with `fail=false`, the test should also pass (testing the server returns the expected input). What we observe is that, the first time `fail=false` is specified, although this test passes, subsequent tests will fail. It's as if the HTTP server created is still around and listening (although that would be odd, because if it were true, subsequent servers should fail to bind to the port).

Anyway, please look at this code - it's a completely reproducible example (I'm using 10.0.0-RC2 and 2.12.0)

package foobar

import java.io.IOException
import java.util.concurrent.atomic.AtomicInteger

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.http.scaladsl.Http
import akka.http.scaladsl.Http.ServerBinding
import akka.http.scaladsl.model.StatusCodes
import akka.http.scaladsl.server.Directives._

import scala.concurrent.duration.Duration
import scala.concurrent.{Await, Future}

class AkkaBugRepr(val data: String) {
def start(port: Int, fail: Boolean): Future[ServerBinding] = {
implicit val system = ActorSystem("bug-repr")
implicit val materializer = ActorMaterializer()

val route = get {
path("status")(if (fail) complete(StatusCodes.NotFound -> "Failing") else complete(data))
}
Http().bindAndHandle(route, java.net.InetAddress.getLocalHost.getHostName, port)
}

}

object ReproduceAkkaBug extends App {
try {

def shortHostname = java.net.InetAddress.getLocalHost.getHostName takeWhile (_ != '.')

val counter = new AtomicInteger()

def runSingleTest(fail: Boolean): Unit = {
val run = counter.incrementAndGet()
val toReturn = s"R$run"
val r = new AkkaBugRepr(toReturn)
val p = 8080
val b = r.start(p, fail = fail)
val serverBinding = Await.result(b, Duration.Inf)
println(s"Run[$run] created server which returns ${if (fail) "404" else toReturn}")
println(s"Run[$run] bound server at $serverBinding")
try {
val src = scala.io.Source.fromURL(s"http://$shortHostname:$p/status")
val ret = src.mkString
try assert(!fail && ret == toReturn, s"Didn't expect: [$ret]")
finally src.close()
}
catch {
case _: IOException if fail => /* fine */
}
finally {
Await.result(serverBinding.unbind(), Duration.Inf)
println(s"Run[$run] unbound server at $serverBinding")

}
println(s"Passed test $run")
println()
}

runSingleTest(fail = true)
runSingleTest(fail = true)
runSingleTest(fail = true)
runSingleTest(fail = false) //this will pass, subsequent tests observe this server "hanging around"
runSingleTest(fail = true)
}
catch {
case t: Throwable => t.printStackTrace(); sys.exit(-1)
}

}