cvasii
commented
10 years ago The client responsible for communicating with the ES cluster was not designed properly. Now it is a singleton with a pool of connections, which should do the trick. I have tested withh 5000 requests in sequence and no expections were thrown
After doing 2703 requests (in sequence, quick one after another) for pages of 1000 passwords, the search application crashes with the following error stack trace:
12:59:49.065 [SearchAkkaSystem-akka.actor.default-dispatcher-5] ERROR akka.actor.OneForOneStrategy - org.apache.http.nio.reactor.IOReactorException: Failure opening selector java.lang.IllegalStateException: org.apache.http.nio.reactor.IOReactorException: Failure opening selector at org.apache.http.impl.nio.client.IOReactorUtils.create(IOReactorUtils.java:43) ~[httpasyncclient-4.0.jar:4.0] at org.apache.http.impl.nio.client.HttpAsyncClientBuilder.build(HttpAsyncClientBuilder.java:603) ~[httpasyncclient-4.0.jar:4.0] at org.apache.http.impl.nio.client.HttpAsyncClients.createDefault(HttpAsyncClients.java:62) ~[httpasyncclient-4.0.jar:4.0] at ro.endava.akka.workshop.es.client.ESRestClientFactory.getClient(ESRestClientFactory.java:45) ~[classes/:na] at ro.endava.akka.workshop.actors.SearchPasswordActor.searchPasswords(SearchPasswordActor.java:41) ~[classes/:na] at ro.endava.akka.workshop.actors.SearchPasswordActor.onReceive(SearchPasswordActor.java:33) ~[classes/:na] at akka.actor.UntypedActor$$anonfun$receive$1.applyOrElse(UntypedActor.scala:167) ~[akka-actor_2.10-2.2.3.jar:2.2.3] at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498) [akka-actor_2.10-2.2.3.jar:2.2.3] at akka.actor.ActorCell.invoke(ActorCell.scala:456) [akka-actor_2.10-2.2.3.jar:2.2.3] at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237) [akka-actor_2.10-2.2.3.jar:2.2.3] at akka.dispatch.Mailbox.run(Mailbox.scala:219) [akka-actor_2.10-2.2.3.jar:2.2.3] at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386) [akka-actor_2.10-2.2.3.jar:2.2.3] at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) [na:na] at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) [na:na] at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) [na:na] at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) [na:na] Caused by: org.apache.http.nio.reactor.IOReactorException: Failure opening selector at org.apache.http.impl.nio.reactor.AbstractMultiworkerIOReactor.(AbstractMultiworkerIOReactor.java:144) ~[httpcore-nio-4.3.2.jar:4.3.2]
at org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor.(DefaultConnectingIOReactor.java:81) ~[httpcore-nio-4.3.2.jar:4.3.2]
at org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor.(DefaultConnectingIOReactor.java:96) ~[httpcore-nio-4.3.2.jar:4.3.2]
at org.apache.http.impl.nio.client.IOReactorUtils.create(IOReactorUtils.java:41) ~[httpasyncclient-4.0.jar:4.0]
... 15 common frames omitted
Caused by: java.io.IOException: Unable to establish loopback connection
at sun.nio.ch.PipeImpl$Initializer.run(Unknown Source) ~[na:1.7.0_17]
at sun.nio.ch.PipeImpl$Initializer.run(Unknown Source) ~[na:1.7.0_17]
at java.security.AccessController.doPrivileged(Native Method) ~[na:1.7.0_17]
at sun.nio.ch.PipeImpl.(Unknown Source) ~[na:1.7.0_17]
at sun.nio.ch.SelectorProviderImpl.openPipe(Unknown Source) ~[na:1.7.0_17]
at java.nio.channels.Pipe.open(Unknown Source) ~[na:1.7.0_17]
at sun.nio.ch.WindowsSelectorImpl.(Unknown Source) ~[na:1.7.0_17]
at sun.nio.ch.WindowsSelectorProvider.openSelector(Unknown Source) ~[na:1.7.0_17]
at java.nio.channels.Selector.open(Unknown Source) ~[na:1.7.0_17]
at org.apache.http.impl.nio.reactor.AbstractMultiworkerIOReactor.(AbstractMultiworkerIOReactor.java:142) ~[httpcore-nio-4.3.2.jar:4.3.2]
... 18 common frames omitted
Caused by: java.net.SocketException: No buffer space available (maximum connections reached?): connect
at sun.nio.ch.Net.connect0(Native Method) ~[na:1.7.0_17]
at sun.nio.ch.Net.connect(Unknown Source) ~[na:1.7.0_17]
at sun.nio.ch.Net.connect(Unknown Source) ~[na:1.7.0_17]
at sun.nio.ch.SocketChannelImpl.connect(Unknown Source) ~[na:1.7.0_17]
at java.nio.channels.SocketChannel.open(Unknown Source) ~[na:1.7.0_17]
... 28 common frames omitted
13:00:09.091 [SearchAkkaSystem-akka.actor.default-dispatcher-2] DEBUG akka.io.TcpListener - New connection accepted
13:00:09.100 [SearchAkkaSystem-akka.actor.default-dispatcher-5] DEBUG s.can.server.HttpServerConnection - Connection was PeerClosed, awaiting TcpConnection termination...
13:00:09.102 [SearchAkkaSystem-akka.actor.default-dispatcher-4] DEBUG s.can.server.HttpServerConnection - TcpConnection terminated, stopping
13:00:10.125 [SearchAkkaSystem-akka.actor.default-dispatcher-4] WARN s.can.server.HttpServerConnection - Configured registration timeout of 1 second expired, stopping
13:00:10.131 [SearchAkkaSystem-akka.actor.default-dispatcher-8] DEBUG akka.io.TcpListener - New connection accepted
13:00:11.150 [SearchAkkaSystem-akka.actor.default-dispatcher-4] WARN s.can.server.HttpServerConnection - Configured registration timeout of 1 second expired, stopping
13:00:11.156 [SearchAkkaSystem-akka.actor.default-dispatcher-9] DEBUG akka.io.TcpListener - New connection accepted
13:00:12.170 [SearchAkkaSystem-akka.actor.default-dispatcher-5] WARN s.can.server.HttpServerConnection - Configured registration timeout of 1 second expired, stopping
13:00:12.176 [SearchAkkaSystem-akka.actor.default-dispatcher-5] DEBUG akka.io.TcpListener - New connection accepted
13:00:13.191 [SearchAkkaSystem-akka.actor.default-dispatcher-9] WARN s.can.server.HttpServerConnection - Configured registration timeout of 1 second expired, stopping
13:00:13.196 [SearchAkkaSystem-akka.actor.default-dispatcher-5] DEBUG akka.io.TcpListener - New connection accepted
13:00:14.209 [SearchAkkaSystem-akka.actor.default-dispatcher-2] WARN s.can.server.HttpServerConnection - Configured registration timeout of 1 second expired, stopping
13:00:38.942 [SearchAkkaSystem-akka.actor.default-dispatcher-5] ERROR akka.actor.RepointableActorRef - Error during processing of request HttpRequest(GET,http://localhost:8080/getPasswords/2703/1000,List(User-Agent: spray-can/1.2.0, Host: localhost:8080),Empty,HTTP/1.1)
java.util.concurrent.TimeoutException: Futures timed out after [50 seconds]
at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:219) ~[na:na]
at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:223) ~[na:na]
at scala.concurrent.Await$$anonfun$result$1.apply(package.scala:107) ~[na:na]
at akka.dispatch.MonitorableThreadFactory$AkkaForkJoinWorkerThread$$anon$3.block(ThreadPoolBuilder.scala:169) ~[akka-actor_2.10-2.2.3.jar:2.2.3]
at scala.concurrent.forkjoin.ForkJoinPool.managedBlock(ForkJoinPool.java:3640) [na:na]
at akka.dispatch.MonitorableThreadFactory$AkkaForkJoinWorkerThread.blockOn(ThreadPoolBuilder.scala:167) ~[akka-actor_2.10-2.2.3.jar:2.2.3]
at scala.concurrent.Await$.result(package.scala:107) ~[na:na]
at rest.ESHttpService$$anonfun$receive$1.apply(ESHttpService.scala:47) ~[classes/:na]
at rest.ESHttpService$$anonfun$receive$1.apply(ESHttpService.scala:42) ~[classes/:na]
at spray.routing.ApplyConverterInstances$$anon$21$$anonfun$apply$2.apply(ApplyConverterInstances.scala:32) ~[spray-routing-1.2.0.jar:na]
at spray.routing.ApplyConverterInstances$$anon$21$$anonfun$apply$2.apply(ApplyConverterInstances.scala:31) ~[spray-routing-1.2.0.jar:na]
at spray.routing.ConjunctionMagnet$$anon$1$$anon$2$$anonfun$happly$1$$anonfun$apply$1.apply(Directive.scala:38) ~[spray-routing-1.2.0.jar:na]
at spray.routing.ConjunctionMagnet$$anon$1$$anon$2$$anonfun$happly$1$$anonfun$apply$1.apply(Directive.scala:37) ~[spray-routing-1.2.0.jar:na]
at spray.routing.directives.BasicDirectives$$anon$1.happly(BasicDirectives.scala:26) ~[spray-routing-1.2.0.jar:na]
at spray.routing.ConjunctionMagnet$$anon$1$$anon$2$$anonfun$happly$1.apply(Directive.scala:37) ~[spray-routing-1.2.0.jar:na]
at spray.routing.ConjunctionMagnet$$anon$1$$anon$2$$anonfun$happly$1.apply(Directive.scala:36) ~[spray-routing-1.2.0.jar:na]
at spray.routing.directives.BasicDirectives$$anon$2.happly(BasicDirectives.scala:79) ~[spray-routing-1.2.0.jar:na]
at spray.routing.ConjunctionMagnet$$anon$1$$anon$2.happly(Directive.scala:36) ~[spray-routing-1.2.0.jar:na]
at spray.routing.Directive$$anon$7$$anonfun$happly$4.apply(Directive.scala:86) ~[spray-routing-1.2.0.jar:na]
at spray.routing.Directive$$anon$7$$anonfun$happly$4.apply(Directive.scala:86) ~[spray-routing-1.2.0.jar:na]
at spray.routing.directives.BasicDirectives$$anon$3$$anonfun$happly$1.apply(BasicDirectives.scala:92) ~[spray-routing-1.2.0.jar:na]
at spray.routing.directives.BasicDirectives$$anon$3$$anonfun$happly$1.apply(BasicDirectives.scala:92) ~[spray-routing-1.2.0.jar:na]
at spray.routing.RouteConcatenation$RouteConcatenation$$anonfun$$tilde$1.apply(RouteConcatenation.scala:30) ~[spray-routing-1.2.0.jar:na]
at spray.routing.RouteConcatenation$RouteConcatenation$$anonfun$$tilde$1.apply(RouteConcatenation.scala:29) ~[spray-routing-1.2.0.jar:na]
at spray.routing.directives.BasicDirectives$$anonfun$mapRequestContext$1$$anonfun$apply$1.apply(BasicDirectives.scala:30) ~[spray-routing-1.2.0.jar:na]
at spray.routing.directives.BasicDirectives$$anonfun$mapRequestContext$1$$anonfun$apply$1.apply(BasicDirectives.scala:30) ~[spray-routing-1.2.0.jar:na]
at spray.routing.directives.ExecutionDirectives$$anonfun$handleExceptions$1$$anonfun$apply$4.apply(ExecutionDirectives.scala:35) ~[spray-routing-1.2.0.jar:na]
at spray.routing.directives.ExecutionDirectives$$anonfun$handleExceptions$1$$anonfun$apply$4.apply(ExecutionDirectives.scala:33) ~[spray-routing-1.2.0.jar:na]
at spray.routing.HttpServiceBase$class.runSealedRoute$1(HttpService.scala:36) ~[spray-routing-1.2.0.jar:na]
at spray.routing.HttpServiceBase$$anonfun$runRoute$1.applyOrElse(HttpService.scala:46) ~[spray-routing-1.2.0.jar:na]
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498) [akka-actor_2.10-2.2.3.jar:2.2.3]
at akka.actor.ActorCell.invoke(ActorCell.scala:456) [akka-actor_2.10-2.2.3.jar:2.2.3]
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237) [akka-actor_2.10-2.2.3.jar:2.2.3]
at akka.dispatch.Mailbox.run(Mailbox.scala:219) [akka-actor_2.10-2.2.3.jar:2.2.3]
at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386) [akka-actor_2.10-2.2.3.jar:2.2.3]
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) [na:na]
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) [na:na]
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) [na:na]
at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) [na:na]