title: exception in vert.x worker thread
body: <p>I'm very new to vert.x platform. I have a standard and a worker verticle in my project which communicates through the eventBus. The worker verticle performs multiple REST API calls in loop and database access.</p>
  <p>My problem is the worker verticle complete the task without issue at some run but sometimes it throws below error.</p>
  <p><code>Exception in thread &quot;vert.x-worker-thread-12&quot; io.vertx.core.VertxException: Connection was closed</code></p>
  <p>I'm using kotlin coroutine to handle <code>constructDevice(vertx: Vertx)</code> function which performs most of the REST API calls and database access.</p>
  <p>Could anyone please tell me what's the reason for the above problem also is there any way to improve <code>constructDevice(vertx: Vertx)</code> function to handle multiple REST API calls and MongoDB access effectively.</p>
  <pre><code>    // worker verticle to handle multiple REST API calls and MongoDB database access
      
      class DeviceDiscoverVerticle : CoroutineVerticle() {
          override suspend fun start() {
              val consumer = vertx.eventBus().localConsumer&lt;String&gt;(&quot;listDevice&quot;)
              consumer.handler { message -&gt;
                  CoroutineScope(vertx.dispatcher()).launch {
                      constructDevice(vertx)
                  }
                  message.reply(&quot;discovered&quot;)
              }
          }
      }
      
      // standard verticle to communicate with worker verticle 
      
      class ListDeviceVerticle : CoroutineVerticle() {
          override suspend fun start() {
              val reply = awaitResult&lt;Message&lt;String&gt;&gt; { h -&gt;
                  vertx.eventBus().request(&quot;listDevice&quot;, &quot;deviceIPs&quot;, h)
              }
              println(&quot;Reply received: ${reply.body()}&quot;)
          }
      }
      
      fun main() {
          val vertx = Vertx.vertx()
          val workOption = DeploymentOptions().setWorker(true)
          vertx.deployVerticle(DeviceDiscoverVerticle(), workOption)
          vertx.deployVerticle(ListDeviceVerticle())
      }
  
  
      suspend fun constructDevice(vertx: Vertx) {
          val deviceRepository = listOf(
              &quot;10.22.0.106&quot;,
              &quot;10.22.0.120&quot;,
              &quot;10.22.0.115&quot;,
              &quot;10.22.0.112&quot;
          )
      
          val webClient = WebClient.create(vertx)
          val config = json { obj(&quot;db_name&quot; to &quot;mnSet&quot;, &quot;connection_string&quot; to &quot;mongodb://localhost:27017&quot;) }
          val mongoClient: MongoClient = MongoClient.create(vertx, config)
          val json = Json(JsonConfiguration.Stable.copy(ignoreUnknownKeys = true))
          
          // loop through the IP list and calls REST endpoints
          
          val deviceList = deviceRepository.map { deviceIP -&gt;
              val deviceIPconfig: DeviceIPconfig
              val deviceType: DeviceType
              val requestDeviceIP: HttpRequest&lt;Buffer&gt; = webClient.get(deviceIP, &quot;emsfp/node/v1/self/ipconfig/&quot;)
              val requestDeviceType: HttpRequest&lt;Buffer&gt; = webClient.get(deviceIP, &quot;emsfp/node/v1/self/information/&quot;)
      
              val responseDeviceIP = awaitResult&lt;HttpResponse&lt;Buffer&gt;&gt; { handler -&gt;
                  requestDeviceIP.send(handler)
              }
              deviceIPconfig = if (responseDeviceIP.statusCode() == 200) {
                  json.parse(DeviceIPconfig.serializer(), responseDeviceIP.bodyAsString())
              } else {
                  println(&quot;request to device $deviceIP failed with ${responseDeviceIP.statusCode()}&quot;)
                  DeviceIPconfig()
              }
              
              val responseDeviceType = awaitResult&lt;HttpResponse&lt;Buffer&gt;&gt; { handler -&gt;
                  requestDeviceType.send(handler)
              }
              if (responseDeviceType.statusCode() == 200) {
                  deviceType = json.parse(DeviceType.serializer(), responseDeviceType.bodyAsString())
                  val device = DeviceModel(deviceIPconfig, deviceType)
                  json {
                      obj(
                          &quot;_id&quot; to deviceIPconfig.localMac,
                          &quot;device&quot; to json.stringify(DeviceModel.serializer(), device)
                      )
                  }
              } else {
                  println(&quot;request to device $deviceIP failed with ${responseDeviceType.statusCode()}&quot;)
                  jsonObjectOf()
              }
      
          }.filterNot { it.isEmpty }
          
          // construct data to upload in mongoDB
          
          val activeDeviceIDs = json {
              obj(&quot;_id&quot; to &quot;activeDeviceIDs&quot;,
                  &quot;activeDeviceIDs&quot; to deviceList.map { it.get&lt;String&gt;(&quot;_id&quot;) })
          }
          val activeDevices = json {
              obj(&quot;_id&quot; to &quot;activeDevices&quot;,
                  &quot;activeDevices&quot; to json { array(deviceList) }
              )
          }
          
          // save the data in MongoDB
          
          mongoClient.save(&quot;devices&quot;, activeDeviceIDs) { res -&gt;
              if (res.succeeded()) {
                  println(&quot;saved successfully&quot;)
              } else {
                  res.cause().printStackTrace()
              }
          }
          mongoClient.save(&quot;devices&quot;, activeDevices) { res -&gt;
              if (res.succeeded()) {
                  println(&quot;saved successfully&quot;)
              } else {
                  res.cause().printStackTrace()
              }
          }
      }
  </code></pre>
  <p><strong>Updated question: 1</strong></p>
  <p>@Damian I have updated my question based on your input. I have simplified my question above for easy to understand, but when I tried to implement the things using promise/future I'm getting stuck at some point.</p>
  <p>My task is to get data from different REST endpoints and contract kotlin class from it and I wants to it in parallel.</p>
  <pre><code>fun constructDeviceDevice(deviceIP: String, device: String, webClient: WebClient): Future&lt;HttpResponse&lt;Buffer&gt;&gt; {
      val requestDevices: HttpRequest&lt;Buffer&gt; = webClient.get(deviceIP, &quot;emsfp/node/v1/self/diag/devices/$device&quot;)
      val deviceDevicePromise: Promise&lt;HttpResponse&lt;Buffer&gt;&gt; = Promise.promise()
  
      requestDevices.send { asyncResult -&gt;
          if (asyncResult.succeeded())
              deviceDevicePromise.complete(asyncResult.result())
          else
              deviceDevicePromise.fail(&quot;Http request failed&quot;);
      }
      return deviceDevicePromise.future()
  }
  
  fun constructDeviceDevices(deviceIP: String, webClient: WebClient): Future&lt;List&lt;Future&lt;HttpResponse&lt;Buffer&gt;&gt;&gt;&gt; {
      val requestDeviceDevices: HttpRequest&lt;Buffer&gt; = webClient.get(deviceIP, &quot;emsfp/node/v1/self/diag/devices/&quot;)
      val deviceDevicesPromise: Promise&lt;List&lt;Future&lt;HttpResponse&lt;Buffer&gt;&gt;&gt;&gt; = Promise.promise()
  
      requestDeviceDevices.send { asyncResult -&gt;
          if (asyncResult.succeeded()) {
              // this will return Json array and each element of that array needs to be called again in a loop.
              val result = asyncResult.result().bodyAsJsonArray().map { device -&gt;
                  constructDeviceDevice(deviceIP, device.toString(), webClient)
              }
              deviceDevicesPromise.complete(result)
          } else
              deviceDevicesPromise.fail(&quot;Http request failed&quot;)
      }
      return deviceDevicesPromise.future()
  }
  
  fun constructDevice(vertx: Vertx, webClient: WebClient, deviceIP: String): List&lt;Future&lt;HttpResponse&lt;Buffer&gt;&gt;&gt; {
  
      val deviceDevicesFuture: Future&lt;List&lt;Future&lt;HttpResponse&lt;Buffer&gt;&gt;&gt;&gt; = constructDeviceDevices(deviceIP, webClient)
      // I need to call other rest points similar to this and I need map the result to kotlin class.
  
     // how do get HTTP response out of each future request in deviceDevicesFuture: Future&lt;List&lt;Future&lt;HttpResponse&lt;Buffer&gt;&gt;&gt;&gt;. 
  
  }
  
  class DeviceDiscoverVerticle : AbstractVerticle() {
          override fun start() {
              val deviceRepository = // list of IP strings
      
              val webClient = WebClient.create(vertx)
              vertx.eventBus().localConsumer&lt;String&gt;(&quot;listDevice&quot;).handler { message -&gt;
                  deviceRepository.forEach { deviceIP -&gt;
                      val futureList = constructDevice(vertx, webClient, deviceIP)
                      CompositeFuture.all(futureList).onComplete { allFuturesResult -&gt;
                              if (allFuturesResult.succeeded()) {
                                  // how to handle individual future result here to construct data
                              } else {
                                  println(&quot;failed&quot;)
                              }
                          }
                  }
              }
          }
  </code></pre>
  <p><strong>Updated question: 2</strong></p>
  <p>@Damian as you suggested I have updated my code.</p>
  <pre><code>fun constructDeviceDevice(deviceIP: String, device: String, webClient: WebClient): Future&lt;HttpResponse&lt;Buffer&gt;&gt; {
      val requestDevices: HttpRequest&lt;Buffer&gt; = webClient.get(deviceIP, &quot;emsfp/node/v1/flows/$device&quot;)
      val deviceDevicePromise: Promise&lt;HttpResponse&lt;Buffer&gt;&gt; = Promise.promise()
  
      requestDevices.send { asyncResult -&gt;
          if (asyncResult.succeeded())
              deviceDevicePromise.complete(asyncResult.result())
          else
              deviceDevicePromise.fail(&quot;Http request failed&quot;)
      }
      return deviceDevicePromise.future()
  }
  
  fun constructDeviceDevices(deviceIP: String, webClient: WebClient): Future&lt;HttpResponse&lt;Buffer&gt;&gt; {
      val requestDeviceDevices: HttpRequest&lt;Buffer&gt; = webClient.get(deviceIP, &quot;emsfp/node/v1/flows/&quot;)
      val deviceDevicesPromise: Promise&lt;HttpResponse&lt;Buffer&gt;&gt; = Promise.promise()
  
      requestDeviceDevices.send { asyncResult -&gt;
          if (asyncResult.succeeded()) {
              deviceDevicesPromise.complete(asyncResult.result())
          }
          else
              deviceDevicesPromise.fail(&quot;Http request failed&quot;)
      }
      return deviceDevicesPromise.future()
  }
  
  
  fun constructDevice(webClient: WebClient, deviceIP: String): Future&lt;DeviceFlow&gt; {
      val json = Json(JsonConfiguration.Stable.copy(ignoreUnknownKeys = true, isLenient = true))
      val constructDevicePromise: Promise&lt;DeviceFlow&gt; = Promise.promise()
      val httpDevicesFuture: Future&lt;HttpResponse&lt;Buffer&gt;&gt; = constructDeviceDevices(deviceIP, webClient)
  
      httpDevicesFuture.onComplete { ar -&gt;
          if(ar.succeeded()) {
              val futureList = ar.result().bodyAsJsonArray().map { device -&gt;
                  constructDeviceDevice(deviceIP, device.toString(), webClient)
              }
              CompositeFuture.all(futureList).onComplete { asyncResult -&gt;
                  if (asyncResult.succeeded()) {
                      asyncResult.result().list&lt;HttpResponse&lt;Buffer&gt;&gt;().forEach { res -&gt;
                          //not all future in futureList are completed here some of them shows Future{unresolved}
                      }
                      constructDevicePromise.complete(DeviceFlow(label = &quot;xyz&quot;))
                  }
                  else {
                      constructDevicePromise.fail(&quot;failed&quot;)
                  }
              }
  
          }
      }
      return constructDevicePromise.future()
  }
  
  
  class DeviceDiscoverVerticle : AbstractVerticle() {
      override fun start() {
          val deviceRepository = //list of IPs
  
          val webClient = WebClient.create(vertx)
          vertx.eventBus().localConsumer&lt;String&gt;(&quot;listDevice&quot;).handler { message -&gt;
              deviceRepository.forEach { deviceIP -&gt;
                  val constructDeviceFuture = constructDevice(webClient, deviceIP)
                  constructDeviceFuture.onComplete {ar -&gt;
                      //println(ar.result().toString())
                  }
              }
          }
      }
  }
  </code></pre>
  <p>My problem is inside</p>
  <pre><code>CompositeFuture.all(futureList).onComplete { asyncResult -&gt;
                          if (asyncResult.succeeded()) {
                              asyncResult.result().list&lt;HttpResponse&lt;Buffer&gt;&gt;().forEach {
  </code></pre>
  <p>here most of the futures are unresolved and the execution gets hanged here.</p>
  <pre><code>[Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@67d2e79}, Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@8bad0c6}, Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@c854509}, Future{unresolved}, Future{unresolved}, Future{cause=Http request failed}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}]
  </code></pre>
  <p>so I have changed <code>CompositeFuture.all(futureList).onComplete</code> to <code>CompositeFuture.join(futureList).onComplete</code> as per vert.x docs join will wait for all future to complete</p>
  <blockquote>
  <p>The join composition waits until all futures are completed, either
  with a success or a failure. CompositeFuture.join takes several
  futures arguments (up to 6) and returns a future that is succeeded
  when all the futures are succeeded and failed when all the futures
  are completed and at least one of them is failed</p>
  </blockquote>
  <p>but now few futures are getting failed. Here is the output of future list after changing to <code>CompositeFuture.join</code></p>
  <pre><code>CompositeFuture.join(futureList).onComplete { asyncResult -&gt;
  println(futureList)
                              if (asyncResult.succeeded()) { res -&gt;
  // println(res) this one gets hanged and not printing all response
                                  asyncResult.result().list&lt;HttpResponse&lt;Buffer&gt;&gt;().forEach {
  
  
  
  [Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@5e9d3832}, Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@379c326a}]
      [Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@51a39962}, Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@edcd528}, Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@293c3e5c}, Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@5f86d3ec}, Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@12a329f7}, Future{cause=Http request failed}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}]
      [Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@7abedb1e}, Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@3238d4cb}, Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@5bc868d3}, Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@50af1ecc}, Future{unresolved}, Future{cause=Http request failed}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}]
      [Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@5cc549ec}, Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@282f4033}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{cause=Http request failed}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}]
      [Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@41a890b3}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{cause=Http request failed}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}]
      [Future{result=io.vertx.ext.web.client.impl.HttpResponseImpl@147d772a}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{cause=Http request failed}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}, Future{unresolved}]
  </code></pre>
  <p>So few futures are getting failed because of my device can't handle the concurrent request? also why the program execution stuck inside</p>
  <pre><code>asyncResult.result().list&lt;HttpResponse&lt;Buffer&gt;&gt;().forEach { 
  </code></pre>
  <p>Incase if It's problem with device concurrency request handling then what would be the other solution for this problem. Is it possible to run this whole rest calls out of vertx environment and communicate with it through event bus?</p>
  <p>Also if I deployed <code>DeviceDiscoverVerticle</code> as standard verticle instead of worker verticle the application gets stuck completely at  <code>CompositeFuture.all(futureList).onComplete</code>.</p>