Connection Leaks :(

[shanielh]

Hi,

I have a program which calls the acquire lock, release lock and delete key operations a lot of times (~ 1 per second).

When I run the command :

watch -n 0.2 "netstat -t -n | grep '127.0.0.1:8500' | wc -l"

In the background, I can see the there are a lot of connections left open to the consul-agent until it reaches my ulimit.

[rickfast]

i'll take a look this weekend

[rickfast]

which JAX-RS implementation are you using?

[shanielh]

I use the Apache one.

[rickfast]

Fixed by be0e212

[rickfast]

I had a few minutes this morning to take a look at this. We had a similar problem with the agent client. I made sure the response is explicitly close. Please try with 0.9.15 and re-open if you have issues.

[shanielh]

I confirm that it works 👍

My test code

import java.util.concurrent.Executors

import com.orbitz.consul.Consul
import com.orbitz.consul.model.session.{ImmutableSession, Session}

import scala.concurrent.{Future, ExecutionContext}

object Program extends App {

  def lockWorker(consul: Consul, workerIndex: Int, sessionId: String) = {
    val kvp = consul.keyValueClient()
    val key = workerIndex.toString


    while (true) {
      val lockResult = kvp.acquireLock(key, sessionId)
      if (lockResult) {
        println(s"Locked $workerIndex")

        Thread sleep 100

        val unlockResult = kvp.releaseLock(key, sessionId)
        if (unlockResult) {
          println(s"Unlocked $workerIndex")
          kvp.deleteKey(key)
        } else {
          println(s"Unable to unlock $workerIndex")
        }
      } else {
        println(s"Unable to lock $workerIndex")
      }
    }

  }


  println("Hello World")
  val consul = Consul.builder().build()

  val sessionName = java.util.UUID.randomUUID().toString
  val sessionRequest = ImmutableSession.builder().name(sessionName).behavior("delete").ttl("60s").build()

  val session = consul.sessionClient().createSession(sessionRequest).getId

  private val numberOfWorkers = 10
  val executionContext = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(numberOfWorkers))

  for (i <- 0 to numberOfWorkers) {
    val workerIndex = i

    Future {
      lockWorker(consul, workerIndex, session)
    }(executionContext)
  }

  println(s"Running $numberOfWorkers in the background, Press enter to exit...")
  scala.io.StdIn.readLine()
}