title: spring xd losing messages when processing huge volume
body: <p>I am using spring xd My stream looks like below and running tests on 3 node container with 1 admin node with rabbit as transport</p>
  
  <pre><code>aws-s3|processor1|http-client|processor2&gt;queue:readyQueue
  </code></pre>
  
  <p>I have created below tap.</p>
  
  <pre><code>tap1  aws-s3&gt;s3Queue
  
  
  tap2  processor1&gt;processorQueue1
  
  tap3  http-client&gt;httpQueue
  </code></pre>
  
  <p>I run below scenarios in my tests:</p>
  
  <p><code>Scenario1</code>: 5 files of 200k =1 Million records
   concurrency of http-client=70  and processor2=30</p>
  
  <p>I see 900k message  s3Queue</p>
  
  <p>I see 889k message  processorQueue1</p>
  
  <p>I see 886k message  httpQueue</p>
  
  <p>I see 883k message  processorQueue2
  Messages are lost everywhere and its random</p>
  
  <pre><code>Scenario2:
  </code></pre>
  
  <p>5 files of 200k =1 Million records and all module concurrency=1</p>
  
  <p>I see 998800 message  s3Queue</p>
  
  <p>I see 998760 message  processorQueue1</p>
  
  <p>I see 997540 message  httpQueue</p>
  
  <p>I see 997530 message  processorQueue2</p>
  
  <p>Even this number is random and not consistent</p>
  
  <pre><code>Scenario3
  </code></pre>
  
  <p>I changed stream as below and concurrency=1  and 5 files of 200k =1 Million records</p>
  
  <pre><code>aws-s3 &gt;testQueue
  </code></pre>
  
  <p>I get all my messages I run 3 times and no issues.I get all my 1 million messages</p>
  
  <pre><code>scenario4
  </code></pre>
  
  <p>I changed stream as below and concurrency=1 5 files of 200k =1 Million records</p>
  
  <pre><code>aws-s3 |processor1 &gt;testQueue2
  </code></pre>
  
  <p>I get all my messages I run 3 times and no issues.I get all my 1 million messages</p>
  
  <p>In scenario4 and scenarion 3 data ingestion is faster and it took 5  min to process 5 million faster and  ingestion was faster in rabbit transport queue like 5k msg per sec</p>
  
  <p>In scenario 1 data ingestion was slower even s3 module was pulling the data very slow like 300 to 1000 msg per sec</p>
  
  <p>In scenario 2 s3 pulled data faster but http client was slow like 100 msg per sec but aws-s3 pulled data fast like  3-4k msg per sec.</p>
  
  <p>I am thinking like seeing xd threading is causing issues and i am losing messages.Please  can you help me how to solve this issue.</p>
  
  <p>update </p>
  
  <pre><code>Scenario 5 
  </code></pre>
  
  <p>I changed <code>reply-timeout</code> to <code>-1</code> in http client and then 
  I lost only 37 msgs  </p>
  
  <p>Now again I run 2nd iteration I lost 25000 msgs i see the bellowing containers log when that happened</p>
  
  <pre><code>2016-03-04T03:42:04-0500 1.2.1.RELEASE ERROR task-scheduler-7 handler.LoggingHandler - org.springframework.messaging.MessageHandlingException: error occurred in message handler [org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint@b6700b1]; nested exception is org.springframework.amqp.AmqpIOException: java.io.IOException
          at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:84)
          at org.springframework.xd.dirt.integration.rabbit.RabbitMessageBus$SendingHandler.handleMessageInternal(RabbitMessageBus.java:891)
          at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:78)
          at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
          at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:101)
          at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:97)
          at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
          at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:287)
          at org.springframework.integration.channel.interceptor.WireTap.preSend(WireTap.java:129)
          at org.springframework.integration.channel.AbstractMessageChannel$ChannelInterceptorList.preSend(AbstractMessageChannel.java:392)
          at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:282)
          at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:245)
          at sun.reflect.GeneratedMethodAccessor204.invoke(Unknown Source)
          at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
          at java.lang.reflect.Method.invoke(Method.java:497)
          at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:317)
          at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190)
          at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157)
          at org.springframework.integration.monitor.DirectChannelMetrics.monitorSend(DirectChannelMetrics.java:114)
          at org.springframework.integration.monitor.DirectChannelMetrics.doInvoke(DirectChannelMetrics.java:98)
          at org.springframework.integration.monitor.DirectChannelMetrics.invoke(DirectChannelMetrics.java:92)
          at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
          at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:207)
          at com.sun.proxy.$Proxy1537.send(Unknown Source)
          at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115)
          at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45)
          at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:95)
          at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:231)
          at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:154)
          at org.springframework.integration.splitter.AbstractMessageSplitter.produceOutput(AbstractMessageSplitter.java:157)
          at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:102)
          at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:105)
  
  Caused by: org.springframework.amqp.AmqpIOException: java.io.IOException
          at org.springframework.amqp.rabbit.support.RabbitExceptionTranslator.convertRabbitAccessException(RabbitExceptionTranslator.java:63)
          at org.springframework.amqp.rabbit.connection.SimpleConnection.createChannel(SimpleConnection.java:51)
          at org.springframework.amqp.rabbit.connection.CachingConnectionFactory$ChannelCachingConnectionProxy.createBareChannel(CachingConnectionFactory.java:758)
          at org.springframework.amqp.rabbit.connection.CachingConnectionFactory$ChannelCachingConnectionProxy.access$300(CachingConnectionFactory.java:747)
          at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.doCreateBareChannel(CachingConnectionFactory.java:419)
          at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.createBareChannel(CachingConnectionFactory.java:395)
          at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.getCachedChannelProxy(CachingConnectionFactory.java:364)
          at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.getChannel(CachingConnectionFactory.java:357)
          at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.access$1100(CachingConnectionFactory.java:75)
          at org.springframework.amqp.rabbit.connection.CachingConnectionFactory$ChannelCachingConnectionProxy.createChannel(CachingConnectionFactory.java:763)
          at org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils$1.createChannel(ConnectionFactoryUtils.java:85)
          at org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils.doGetTransactionalResourceHolder(ConnectionFactoryUtils.java:134)
          at org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils.getTransactionalResourceHolder(ConnectionFactoryUtils.java:67)
          at org.springframework.amqp.rabbit.core.RabbitTemplate.doExecute(RabbitTemplate.java:1035)
          at org.springframework.amqp.rabbit.core.RabbitTemplate.execute(RabbitTemplate.java:1028)
          at org.springframework.amqp.rabbit.core.RabbitTemplate.send(RabbitTemplate.java:540)
          at org.springframework.amqp.rabbit.core.RabbitTemplate.convertAndSend(RabbitTemplate.java:635)
          at org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint.send(AmqpOutboundEndpoint.java:331)
          at org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint.handleRequestMessage(AmqpOutboundEndpoint.java:323)
          at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:99)
          at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:78)
          ... 93 more
  Caused by: java.io.IOException
          at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:106)
          at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:102)
          at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:124)
          at com.rabbitmq.client.impl.ChannelN.open(ChannelN.java:125)
          at com.rabbitmq.client.impl.ChannelManager.createChannel(ChannelManager.java:134)
          at com.rabbitmq.client.impl.AMQConnection.createChannel(AMQConnection.java:499)
          at org.springframework.amqp.rabbit.connection.SimpleConnection.createChannel(SimpleConnection.java:44)
          ... 112 more
  Caused by: com.rabbitmq.client.ShutdownSignalException: connection error
          at com.rabbitmq.utility.ValueOrException.getValue(ValueOrException.java:67)
          at com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue(BlockingValueOrException.java:33)
          at com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply(AMQChannel.java:348)
          at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:221)
          at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:118)
          ... 116 more
  Caused by: com.rabbitmq.client.impl.UnknownChannelException: Unknown channel number 23364
          at com.rabbitmq.client.impl.ChannelManager.getChannel(ChannelManager.java:80)
          at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:552)
          ... 1 more
  
  2016-03-04T03:42:05-0500 1.2.1.RELEASE ERROR AMQP Connection xxx:5672 connection.CachingConnectionFactory - Channel shutdown: channel error; protocol method: #method&lt;channel.close&gt;(reply-code=404, reply-text=NOT_FOUND - no queue 'xdbus.tap-s3.tap:stream:stream.batch-aws-s3-source.0' in vhost '/', class-id=50, method-id=20)
  
  
  2016-03-04T03:53:13-0500 1.2.1.RELEASE ERROR AMQP Connection xxx:5672 connection.CachingConnectionFactory - Channel shutdown: connection error
  2016-03-04T03:53:13-0500 1.2.1.RELEASE ERROR AMQP Connection xxx:5672 connection.CachingConnectionFactory - Channel shutdown: channel error; protocol method: #method&lt;channel.close&gt;(reply-code=404, reply-text=NOT_FOUND - no queue 'xdbus.tap-s3.tap:stream:stream.batch-aws-s3-source.0' in vhost '/', class-id=50, method-id=20)
  ~                                                                                                                                                                                 
  
  
  2016-03-04T02:57:54-0500 1.2.1.RELEASE ERROR AMQP Connection xxx:8080 connection.CachingConnectionFactory - Channel shutdown: connection error
  2016-03-04T02:57:55-0500 1.2.1.RELEASE ERROR AMQP Connection xxx:8080 connection.CachingConnectionFactory - Channel shutdown: connection error
  2016-03-04T03:42:04-0500 1.2.1.RELEASE ERROR AMQP Connection yyy:5672 connection.CachingConnectionFactory - Channel shutdown: connection error
  </code></pre>
  
  <p>Updated  </p>
  
  <p>I found the issue for message loses when this exception happens i see lot of msg lost.This pattern i tested multiple time.Everytime this exception happens i see msg lost.Also bumping up concurrency makes this issue to occur often.</p>
  
  <pre><code>2016-03-05T13:59:41-0500 1.2.1.RELEASE ERROR AMQP Connection host1:5672 connection.CachingConnectionFactory - Channel shutdown: connection error
  </code></pre>
  
  <p>rabbit configuration</p>
  
  <pre><code>spring:
    rabbitmq:
     addresses: host1:5672,host2:5672,host3:5672
     adminAddresses: http://host1:15672,http://host2:15672,http://host3:15672
     nodes: rabbit@host1.test.com,rabbit@host2.test.com,rabbit@host2.test.com
     username: test
     password: test
     virtual_host: /
     useSSL: false
     sslProperties:
  </code></pre>
  
  <p>updated with increasing cache size to 200</p>
  
  <p>I added xml provided by you and increased cache size to 200.This is the  way happens when processing 1 million and  80 k  messages.Only my http client concurrency is 100 all other is 1 .Slowly processing stopped msg are still there before http-client queue and  same count.But msg count in my named channel slowly increasing like 10 msg per minute but its very slow 
  s3-poller|processor|http-client>queue:batchCacheQueue</p>
  
  <p>Msg not getting decreass  in queue before http  186174.But slowly msg are coming in to batchCacheQueue</p>
  
  <p><a href="https://i.stack.imgur.com/SdoaV.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/SdoaV.png" alt="enter image description here"></a></p>
  
  <p>Test case to simulate:</p>
  
  <p>1)I was using spring integration aws-s3 source with a splitter in composite module | processor like xml parsing |http-client with concurrency 100 >named channel.</p>
  
  <p>2)I think file source might also work.Create single file of million records and try to   pull this from file.</p>
  
  <p>3)After some 4 to 5 run we see this exception happening </p>