title: Spring integration - Queue/Poller seems to exhaust threadpool without any action
body: <p>I have a Spring integration app, attached to an AMQP broker.</p>
  
  <p>I want to receive messages from an amqp-queue, and update db records.</p>
  
  <p>In order to improve performance, I have a pool of workers allowing multiple updates to occur concurrently.</p>
  
  <p>I have the following configuration:</p>
  
  <pre><code>&lt;int-amqp:inbound-channel-adapter queue-names="pricehub.fixtures.priceUpdates.queue" 
                              channel="pricehub.fixtures.priceUpdates.channel"
                              message-converter="jsonMessageConverter"/&gt;
  
  &lt;int:channel id="pricehub.fixtures.priceUpdates.channel"&gt;
      &lt;int:queue  /&gt;
  &lt;/int:channel&gt;
  
  &lt;int:service-activator ref="updatePriceAction" 
       method="updatePrices" 
       input-channel="pricehub.instruments.priceUpdates.channel"&gt;
      &lt;int:poller fixed-delay="50" time-unit="MILLISECONDS" task-executor="taskExecutor" /&gt;
  &lt;/int:service-activator&gt;
  
  &lt;task:executor id="taskExecutor" pool-size="5-50" keep-alive="120" queue-capacity="500"/&gt;
  </code></pre>
  
  <p>If I start this running, with no inbound messages to process on the AMQP channel, I quickly see the thredpool get exhausted, and start rejecting.</p>
  
  <p>Here's the logs:</p>
  
  <pre><code>[Thu Apr 2013 23:41:51.153] DEBUG [] (org.springframework.amqp.rabbit.listener.BlockingQueueConsumer:185) - Retrieving delivery for Consumer: tag=[amq.ctag-w4qPp60jVEQOIEovR4cERv], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,1), acknowledgeMode=AUTO local queue size=0
  [Thu Apr 2013 23:41:51.160] DEBUG [] (org.springframework.amqp.rabbit.listener.BlockingQueueConsumer:185) - Retrieving delivery for Consumer: tag=[amq.ctag-Q3Lq4R9g9E8WBNVLYzaFmq], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,2), acknowledgeMode=AUTO local queue size=0
  [Thu Apr 2013 23:41:51.166] DEBUG [] (org.springframework.amqp.rabbit.listener.BlockingQueueConsumer:185) - Retrieving delivery for Consumer: tag=[amq.ctag-w8bg7ltEV2mot8QXDPCmfK], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,3), acknowledgeMode=AUTO local queue size=0
  [Thu Apr 2013 23:41:51.170] DEBUG [] (org.springframework.amqp.rabbit.listener.BlockingQueueConsumer:185) - Retrieving delivery for Consumer: tag=[amq.ctag-A-0KdqhFjpc-Hvjmv7aZAc], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,4), acknowledgeMode=AUTO local queue size=0
  [Thu Apr 2013 23:41:51.180] DEBUG [] (org.springframework.integration.endpoint.PollingConsumer:71) - Received no Message during the poll, returning 'false'
  [Thu Apr 2013 23:41:51.180] DEBUG [] (org.springframework.integration.endpoint.PollingConsumer:71) - Received no Message during the poll, returning 'false'
  [Thu Apr 2013 23:41:51.199] DEBUG [] (org.springframework.integration.endpoint.PollingConsumer:71) - Received no Message during the poll, returning 'false'
  [Thu Apr 2013 23:41:51.200] DEBUG [] (org.springframework.integration.endpoint.PollingConsumer:71) - Received no Message during the poll, returning 'false'
  [Thu Apr 2013 23:41:51.220] DEBUG [] (org.springframework.integration.endpoint.PollingConsumer:71) - Received no Message during the poll, returning 'false'
  </code></pre>
  
  <p>Pretty quickly, the thread pool starts rejecting the executions:</p>
  
  <pre><code>[Thu Apr 2013 23:47:15.363] ERROR [] (org.springframework.integration.handler.LoggingHandler:126) - org.springframework.core.task.TaskRejectedException: Executor [java.util.concurrent.ThreadPoolExecutor@6ff3cb0e] did not accept task: org.springframework.integration.util.ErrorHandlingTaskExecutor$1@78615c8b
      at org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.execute(ThreadPoolTaskExecutor.java:244)
      at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:49)
      at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller.run(AbstractPollingEndpoint.java:231)
      at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:53)
      at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:81)
      at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:439)
      at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)
      at java.util.concurrent.FutureTask.run(FutureTask.java:138)
      at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:98)
      at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:206)
      at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
      at java.lang.Thread.run(Thread.java:680)
  Caused by: java.util.concurrent.RejectedExecutionException
      at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:1768)
      at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:767)
      at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:658)
      at org.springframework.sched
  </code></pre>
  
  <p>uling.concurrent.ThreadPoolTaskExecutor.execute(ThreadPoolTaskExecutor.java:241)
      ... 12 more</p>
  
  <p>I suspect that the culprit lies here: <code>BlockingQueueConsumer</code> - indicating that each poll for a message blocks the thread until a message arrives ... leading to the threadpool being exhausted quickly.</p>
  
  <p>What's the correct way to configure this?</p>