rabbitmq listeners are not consuming after connection reset

[zvikai]

We have 2 listeners (java processes inside docker on separated machines) for a rabbit queue(on cloudamqp) that were reading from a queue, and then for some reason stopped reading.
According to our logs, on September 30th 2018, 11:28:37.848 (UTC) our application experienced a connection reset:

  September 30th 2018, 11:28:37.848 impl.ForgivingExceptionHandler (ForgivingExceptionHandler.java:120) - An unexpected connection driver error occurred (Exception message: Connection reset)
September 30th 2018, 11:28:37.849 connection.CachingConnectionFactory$DefaultChannelCloseLogger (CachingConnectionFactory.java:1278) - Channel shutdown: connection error
September 30th 2018, 11:28:37.849 connection.CachingConnectionFactory$DefaultChannelCloseLogger (CachingConnectionFactory.java:1278) - Channel shutdown: connection error
September 30th 2018, 11:28:39.401 connection.AbstractConnectionFactory (AbstractConnectionFactory.java:360) - Created new connection: connectionFactory#623d41c5:2/SimpleConnection@1beb3736 [delegate=amqp://xxxx@54.86.215.255:5672/prod, localPort= 57360]  

It seems that since the connection was reset and restored, but we are not getting any new messages from rabbit since then, although there are many messages there.
In RabbitMQ Manager I can see the queue size is getting larger but there are no consumers.
On thread dump I can see the AMQP thread is running.
On netstat I can see connection to cloudamqp is established.
 
We are running a java application with:
openjdk version "1.8.0_151"
org.springframework.boot:spring-boot-starter-amqp:2.0.4.RELEASE
 
Any help on how to continue investigating.
We don’t want to restart the dockers until it is investigated.

Thanks

[garyrussell]

Please show the thread dump and logs.

[zvikai]

repeating log:

listener.BlockingQueueConsumer (BlockingQueueConsumer.java:484) - Retrieving delivery for Consumer@47629063: tags=[{amq.ctag-w2cwGtZZ1oP_T5WNzfVS5A=segment.router}], channel=Cached Rabbit Channel: AMQChannel(amqp://zgwcvcgk@34.229.170.9:5672/prod,1), conn: Proxy@6a5fad0a Shared Rabbit Connection: SimpleConnection@1e8f91a [delegate=amqp://xxxxxx@34.229.170.9:5672/prod, localPort= 56426], acknowledgeMode=NONE local queue size=0

thread dump:

2018-10-03 18:50:06
Full thread dump OpenJDK 64-Bit Server VM (25.151-b12 mixed mode):

"Attach Listener" #43946 daemon prio=9 os_prio=0 tid=0x000055f0a1dfe000 nid=0x2da9 waiting on condition [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE

   Locked ownable synchronizers:
        - None

"pool-2-thread-13" #43936 prio=5 os_prio=0 tid=0x000055f09de91000 nid=0x2cdd waiting on condition [0x00007f389e2c2000]
   java.lang.Thread.State: TIMED_WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000f1e070e8> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
        at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
        at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
        at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"AMQP Connection 34.229.170.9:5672" #43935 prio=5 os_prio=0 tid=0x000055f0a0472000 nid=0x2cdc runnable [0x00007f389f6ce000]
   java.lang.Thread.State: RUNNABLE
        at java.net.SocketInputStream.socketRead0(Native Method)
        at java.net.SocketInputStream.socketRead(SocketInputStream.java:116)
        at java.net.SocketInputStream.read(SocketInputStream.java:171)
        at java.net.SocketInputStream.read(SocketInputStream.java:141)
        at java.io.BufferedInputStream.fill(BufferedInputStream.java:246)
        at java.io.BufferedInputStream.read(BufferedInputStream.java:265)
        - locked <0x00000000f1e06c30> (a java.io.BufferedInputStream)
        at java.io.DataInputStream.readUnsignedByte(DataInputStream.java:288)
        at com.rabbitmq.client.impl.Frame.readFrom(Frame.java:91)
        at com.rabbitmq.client.impl.SocketFrameHandler.readFrame(SocketFrameHandler.java:164)
        - locked <0x00000000f1e06c10> (a java.io.DataInputStream)
        at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:571)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"Scheduler-675100200" #47 prio=5 os_prio=0 tid=0x000055f0a1824000 nid=0x40 waiting on condition [0x00007f389e9c3000]
   java.lang.Thread.State: TIMED_WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000f1b830b8> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
        at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
        at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
        at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"DestroyJavaVM" #36 prio=5 os_prio=0 tid=0x000055f0a0553800 nid=0x12 waiting on condition [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE

   Locked ownable synchronizers:
        - None

"SimpleAsyncTaskExecutor-1" #31 prio=5 os_prio=0 tid=0x000055f09fc8b000 nid=0x32 waiting on condition [0x00007f389f7cf000]
   java.lang.Thread.State: TIMED_WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000f1bb5220> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
        at java.util.concurrent.LinkedBlockingQueue.poll(LinkedBlockingQueue.java:467)
        at org.springframework.amqp.rabbit.listener.BlockingQueueConsumer.nextMessage(BlockingQueueConsumer.java:490)
        at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1249)
        at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:1224)
        at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:102)
        at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1470)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"spring.cloud.inetutils" #30 daemon prio=5 os_prio=0 tid=0x000055f09f934800 nid=0x31 waiting on condition [0x00007f389fad0000]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000f1741610> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
        at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)
        at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"AsyncReporter(org.springframework.cloud.sleuth.zipkin2.sender.RestTemplateSender@7c857e8f)" #29 daemon prio=5 os_prio=0 tid=0x000055f09feec000 nid=0x30 waiting on condition [0x00007f38a0158000]
   java.lang.Thread.State: TIMED_WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000f1915678> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
        at zipkin2.reporter.ByteBoundedQueue.drainTo(ByteBoundedQueue.java:83)
        at zipkin2.reporter.AsyncReporter$BoundedAsyncReporter.flush(AsyncReporter.java:238)
        at zipkin2.reporter.AsyncReporter$Builder.lambda$build$0(AsyncReporter.java:178)
        at zipkin2.reporter.AsyncReporter$Builder$$Lambda$1.run(Unknown Source)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"java-sdk-http-connection-reaper" #28 daemon prio=5 os_prio=0 tid=0x000055f0a12c2000 nid=0x2f waiting on condition [0x00007f389fdd1000]
   java.lang.Thread.State: TIMED_WAITING (sleeping)
        at java.lang.Thread.sleep(Native Method)
        at com.amazonaws.http.IdleConnectionReaper.run(IdleConnectionReaper.java:181)

   Locked ownable synchronizers:
        - None

"Scheduler-1305596603" #26 prio=5 os_prio=0 tid=0x000055f0a18b8800 nid=0x2e waiting on condition [0x00007f38a0914000]
   java.lang.Thread.State: TIMED_WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000f10bcbb8> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
        at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
        at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
        at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"qtp663741480-24" #24 prio=5 os_prio=0 tid=0x000055f09fd62000 nid=0x2d waiting on condition [0x00007f38a0a15000]
   java.lang.Thread.State: TIMED_WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000f0efa2b8> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
        at org.eclipse.jetty.util.BlockingArrayQueue.poll(BlockingArrayQueue.java:392)
        at org.eclipse.jetty.util.thread.QueuedThreadPool.idleJobPoll(QueuedThreadPool.java:571)
        at org.eclipse.jetty.util.thread.QueuedThreadPool.access$800(QueuedThreadPool.java:50)
        at org.eclipse.jetty.util.thread.QueuedThreadPool$2.run(QueuedThreadPool.java:634)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"qtp663741480-23" #23 prio=5 os_prio=0 tid=0x000055f09fd5f800 nid=0x2c waiting on condition [0x00007f38a0b16000]
   java.lang.Thread.State: TIMED_WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000f0efa2b8> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
        at org.eclipse.jetty.util.BlockingArrayQueue.poll(BlockingArrayQueue.java:392)
        at org.eclipse.jetty.util.thread.QueuedThreadPool.idleJobPoll(QueuedThreadPool.java:571)
        at org.eclipse.jetty.util.thread.QueuedThreadPool.access$800(QueuedThreadPool.java:50)
        at org.eclipse.jetty.util.thread.QueuedThreadPool$2.run(QueuedThreadPool.java:634)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"qtp663741480-22" #22 prio=5 os_prio=0 tid=0x000055f09fd5d800 nid=0x2b waiting on condition [0x00007f38a0c17000]
   java.lang.Thread.State: TIMED_WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000f0efa2b8> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
        at org.eclipse.jetty.util.BlockingArrayQueue.poll(BlockingArrayQueue.java:392)
        at org.eclipse.jetty.util.thread.QueuedThreadPool.idleJobPoll(QueuedThreadPool.java:571)
        at org.eclipse.jetty.util.thread.QueuedThreadPool.access$800(QueuedThreadPool.java:50)
        at org.eclipse.jetty.util.thread.QueuedThreadPool$2.run(QueuedThreadPool.java:634)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"qtp663741480-21" #21 prio=5 os_prio=0 tid=0x000055f09e6c1800 nid=0x2a waiting on condition [0x00007f38a0fad000]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000f1c33fe8> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
        at org.eclipse.jetty.util.thread.ReservedThreadExecutor$ReservedThread.reservedWait(ReservedThreadExecutor.java:193)
        at org.eclipse.jetty.util.thread.ReservedThreadExecutor$ReservedThread.run(ReservedThreadExecutor.java:225)
        at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:679)
        at org.eclipse.jetty.util.thread.QueuedThreadPool$2.run(QueuedThreadPool.java:597)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"qtp663741480-20" #20 prio=5 os_prio=0 tid=0x000055f0a09dc000 nid=0x29 waiting on condition [0x00007f38a10ae000]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000f1c4a758> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
        at org.eclipse.jetty.util.thread.ReservedThreadExecutor$ReservedThread.reservedWait(ReservedThreadExecutor.java:193)
        at org.eclipse.jetty.util.thread.ReservedThreadExecutor$ReservedThread.run(ReservedThreadExecutor.java:225)
        at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:679)
        at org.eclipse.jetty.util.thread.QueuedThreadPool$2.run(QueuedThreadPool.java:597)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"qtp663741480-19" #19 prio=5 os_prio=0 tid=0x000055f0a2019800 nid=0x28 waiting on condition [0x00007f38a13fc000]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000f1c38aa0> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
        at org.eclipse.jetty.util.thread.ReservedThreadExecutor$ReservedThread.reservedWait(ReservedThreadExecutor.java:193)
        at org.eclipse.jetty.util.thread.ReservedThreadExecutor$ReservedThread.run(ReservedThreadExecutor.java:225)
        at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:679)
        at org.eclipse.jetty.util.thread.QueuedThreadPool$2.run(QueuedThreadPool.java:597)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"qtp663741480-18-acceptor-0@5181107b-ServerConnector@4fee14b{HTTP/1.1,[http/1.1]}{0.0.0.0:8080}" #18 prio=3 os_prio=0 tid=0x000055f0a2018800 nid=0x27 runnable [0x00007f38a14fd000]
   java.lang.Thread.State: RUNNABLE
        at sun.nio.ch.ServerSocketChannelImpl.accept0(Native Method)
        at sun.nio.ch.ServerSocketChannelImpl.accept(ServerSocketChannelImpl.java:422)
        at sun.nio.ch.ServerSocketChannelImpl.accept(ServerSocketChannelImpl.java:250)
        - locked <0x00000000f1b82608> (a java.lang.Object)
        at org.eclipse.jetty.server.ServerConnector.accept(ServerConnector.java:379)
        at org.eclipse.jetty.server.AbstractConnector$Acceptor.run(AbstractConnector.java:643)
        at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:679)
        at org.eclipse.jetty.util.thread.QueuedThreadPool$2.run(QueuedThreadPool.java:597)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"qtp663741480-17" #17 prio=5 os_prio=0 tid=0x000055f0a201c800 nid=0x26 runnable [0x00007f38a2644000]
   java.lang.Thread.State: RUNNABLE
        at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
        at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
        at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
        at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
        - locked <0x00000000f1b81c08> (a sun.nio.ch.Util$3)
        - locked <0x00000000f1b82160> (a java.util.Collections$UnmodifiableSet)
        - locked <0x00000000f1b82018> (a sun.nio.ch.EPollSelectorImpl)
        at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
        at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:101)
        at org.eclipse.jetty.io.ManagedSelector$SelectorProducer.select(ManagedSelector.java:346)
        at org.eclipse.jetty.io.ManagedSelector$SelectorProducer.produce(ManagedSelector.java:299)
        at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.doProduce(EatWhatYouKill.java:179)
        at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.produce(EatWhatYouKill.java:140)
        at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.run(EatWhatYouKill.java:131)
        at org.eclipse.jetty.util.thread.ReservedThreadExecutor$ReservedThread.run(ReservedThreadExecutor.java:243)
        at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:679)
        at org.eclipse.jetty.util.thread.QueuedThreadPool$2.run(QueuedThreadPool.java:597)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"Spring-Cloud-Vault-2" #15 daemon prio=5 os_prio=0 tid=0x000055f0a123a000 nid=0x24 waiting on condition [0x00007f38a2e43000]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000f07c55f0> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
        at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
        at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
        at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"Spring-Cloud-Vault-1" #14 daemon prio=5 os_prio=0 tid=0x000055f0a1238800 nid=0x23 waiting on condition [0x00007f38a2f44000]
   java.lang.Thread.State: TIMED_WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000000f07c55f0> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
        at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2078)
        at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1093)
        at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
        at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None

"Service Thread" #8 daemon prio=9 os_prio=0 tid=0x000055f09dd05000 nid=0x1e runnable [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE

   Locked ownable synchronizers:
        - None

"C1 CompilerThread2" #7 daemon prio=9 os_prio=0 tid=0x000055f09dd01000 nid=0x1d waiting on condition [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE

   Locked ownable synchronizers:
        - None

"C2 CompilerThread1" #6 daemon prio=9 os_prio=0 tid=0x000055f09dcff000 nid=0x1c waiting on condition [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE

   Locked ownable synchronizers:
        - None

"C2 CompilerThread0" #5 daemon prio=9 os_prio=0 tid=0x000055f09dcfc000 nid=0x1b waiting on condition [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE

   Locked ownable synchronizers:
        - None

"Signal Dispatcher" #4 daemon prio=9 os_prio=0 tid=0x000055f09dcf9800 nid=0x1a runnable [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE

   Locked ownable synchronizers:
        - None

"Finalizer" #3 daemon prio=8 os_prio=0 tid=0x000055f09dccc000 nid=0x19 in Object.wait() [0x00007f38a8848000]
   java.lang.Thread.State: WAITING (on object monitor)
        at java.lang.Object.wait(Native Method)
        at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:143)
        - locked <0x00000000f0278a30> (a java.lang.ref.ReferenceQueue$Lock)
        at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:164)
        at java.lang.ref.Finalizer$FinalizerThread.run(Finalizer.java:209)

   Locked ownable synchronizers:
        - None

"Reference Handler" #2 daemon prio=10 os_prio=0 tid=0x000055f09dcc9800 nid=0x18 in Object.wait() [0x00007f38a8949000]
   java.lang.Thread.State: WAITING (on object monitor)
        at java.lang.Object.wait(Native Method)
        at java.lang.Object.wait(Object.java:502)
        at java.lang.ref.Reference.tryHandlePending(Reference.java:191)
        - locked <0x00000000f0167120> (a java.lang.ref.Reference$Lock)
        at java.lang.ref.Reference$ReferenceHandler.run(Reference.java:153)

   Locked ownable synchronizers:
        - None

"VM Thread" os_prio=0 tid=0x000055f09dcbf000 nid=0x17 runnable 

"GC task thread#0 (ParallelGC)" os_prio=0 tid=0x000055f09db8f800 nid=0x13 runnable 

"GC task thread#1 (ParallelGC)" os_prio=0 tid=0x000055f09db91000 nid=0x14 runnable 

"GC task thread#2 (ParallelGC)" os_prio=0 tid=0x000055f09db93000 nid=0x15 runnable 

"GC task thread#3 (ParallelGC)" os_prio=0 tid=0x000055f09db94800 nid=0x16 runnable 

"VM Periodic Task Thread" os_prio=0 tid=0x000055f09dd9b800 nid=0x1f waiting on condition 

JNI global references: 400

[artembilan]

The Retrieving delivery for Consumer means that you definitely consume message from the queue. Or the problem is not related to what you have just shared or you share logs and thread dump in the non-relevant state of your application.

I don't see anything obvious unless you enable native RabbitMQ Cleint recovery via:

	/**
	 * Set to true to enable amqp-client automatic recovery. Note: Spring AMQP
	 * implements its own connection recovery and this is generally not needed.
	 * @param automaticRecoveryEnabled true to enable.
	 * @since 1.7.1
	 */
	public void setAutomaticRecoveryEnabled(boolean automaticRecoveryEnabled) {

[garyrussell]

@artembilan That trace message is just for polling the internal queue; it doesn't mean we have actually received any messages.

In RabbitMQ Manager I can see the queue size is getting larger but there are no consumers.

tags=[{amq.ctag-w2cwGtZZ1oP_T5WNzfVS5A=segment.router}]

If you don't see a consumer with this tag on the Admin UI then it means the connection was silently dropped somehow which shouldn't be possible, unless you have disabled heartbeats.

[slevinson]

Hey,

we don't see any consumers for the queue,
but the guys at cloudamqp helped us identify the connection in Admin UI
These are the connection details and client properties, seems that heartbeat is enabled.
Please let me know if any additional info could help.

Details
Node rabbit@*********
Client-provided name connectionFactory#623d41c5:2
Username *********
Protocol AMQP 0-9-1
Connected at 2018-09-30 14:28:39
Authentication PLAIN
State running
Heartbeat 60s
Frame max 131072 bytes
Channel limit 0 channels

Client properties
connection_name connectionFactory#623d41c5:2
product RabbitMQ
copyright Copyright (c) 2007-2016 Pivotal Software, Inc.
capabilities
exchange_exchange_bindings: true
connection.blocked: true
authentication_failure_close: true
basic.nack: true
publisher_confirms: true
consumer_cancel_notify: true
information Licensed under the MPL. See http://www.rabbitmq.com/
version 4.0.3
platform Java

[garyrussell]

We need to see the logs at the time the connection failure.

[slevinson]

Here are the logs, unfortunately we had it on INFO level,
and there are mostly application logs.
You can see the connection error log at 11:28:37.848
download link

[slevinson]

after restarting the service, we saw the same behaviour again.
here are logs with debug level. connection reset at 2018-10-04 08:54:02.201
download link

[garyrussell]

In the second log, I see messages continue to be received after the connection reset.

DEBUG | 2018-10-04 08:54:06.157 | [SimpleAsyncTaskExecutor-1] |  |  | listener.BlockingQueueConsumer (BlockingQueueConsumer.java:484) - Retrieving delivery for Consumer@5183131d: tags=[{amq.ctag-e4PckB8PDr8bKh9pnM2DRA=segment.router}], channel=Cached Rabbit Channel: AMQChannel(amqp://xxxxxxxx@34.229.170.9:5672/prod,1), conn: Proxy@60cae36 Shared Rabbit Connection: SimpleConnection@1c6eb78d [delegate=amqp://xxxxxxxx@34.229.170.9:5672/prod, localPort= 35534], acknowledgeMode=NONE local queue size=1
DEBUG | 2018-10-04 08:54:06.157 | [SimpleAsyncTaskExecutor-1] |  |  | listener.BlockingQueueConsumer (BlockingQueueConsumer.java:454) - Received message: (Body:'[B@1aa3fbfa(byte[285])' MessageProperties [headers={breadcrumbId=ID-240a0d3530bc-1538225822584-0-45903, Content-Type=application/json, Segment-Type=AUDIO}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=null, contentEncoding=null, contentLength=0, deliveryMode=null, receivedDeliveryMode=null, expiration=null, priority=null, redelivered=false, receivedExchange=segment.exchange, receivedRoutingKey=, receivedDelay=null, deliveryTag=11, messageCount=0, consumerTag=amq.ctag-e4PckB8PDr8bKh9pnM2DRA, consumerQueue=segment.router])
DEBUG | 2018-10-04 08:54:06.157 | [pool-2-thread-4] |  |  | listener.BlockingQueueConsumer$InternalConsumer (BlockingQueueConsumer.java:865) - Storing delivery for Consumer@5183131d: tags=[{amq.ctag-e4PckB8PDr8bKh9pnM2DRA=segment.router}], channel=Cached Rabbit Channel: AMQChannel(amqp://xxxxxxxx@34.229.170.9:5672/prod,1), conn: Proxy@60cae36 Shared Rabbit Connection: SimpleConnection@1c6eb78d [delegate=amqp://xxxxxxxx@34.229.170.9:5672/prod, localPort= 35534], acknowledgeMode=NONE local queue size=1

The connection from port 35534 doesn't appear to have been broken; it was opened at 08:53:16.979.

The connection that broke at 08:54:02.201 appears to be a different connection (health indicator?).

In the first log, it is rather strange that there are no error logs from the listener container (or BlockingQueueConsumer).

However, the SimpleAsyncTaskExecutor-2 thread stopped logging at 11:29:05.671.

That thread does not appear in your thread dump above; and SimpleAsyncTaskExecutor-1 in your thread dump does not appear in the log, so these two artifacts don't appear to reflect the same environment.

[slevinson]

Thanks for your response, and sorry for the confusion,
those are indeed two individual environments, both had the same behaviour at about the same time.
Here is the log that corresponds to SimpleAsyncTaskExecutor-1
connection reset at 2018-09-30 11:25:57.396
download link

[slevinson]

It is indeed weird that the listener was still handling messages, I figured it was just emptying
the local queue.

The answer that we got from cloudamqp was that the connection was reset due to a timeout.
At the same time the consumers were removed from the UI, and after the connection was reset,
the consumers never came back.

[garyrussell]

Very odd; in both cases, I see the connection reset but no errors in the containers. I do see 9 messsages processed after the failure - presumably you have a prefetch (maybe 10?).

So it simply appears that the container is not informed about the connection loss.

In both cases, however, the new connection is created on threads named qtp1418129042-20 and qtp663741480-17.

The normal behavior is

2018-10-04 13:30:33.923 ERROR 98685 --- [ 127.0.0.1:5672] o.s.a.r.c.CachingConnectionFactory       : Channel shutdown: connection error; protocol method: #method<connection.close>(reply-code=320, reply-text=CONNECTION_FORCED - broker forced connection closure with reason 'shutdown', class-id=0, method-id=0)
2018-10-04 13:30:33.925  WARN 98685 --- [ 127.0.0.1:5672] c.r.c.impl.ForgivingExceptionHandler     : An unexpected connection driver error occured (Exception message: Connection reset)
2018-10-04 13:30:34.516  INFO 98685 --- [cTaskExecutor-1] o.s.a.r.l.SimpleMessageListenerContainer : Restarting Consumer@2371aaca: tags=[{amq.ctag-pHi82CA8JHilDvCpjRV-Uw=spring.gen-ILh4GFNDRtS232ZJcrjs9g}], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,1), conn: Proxy@63e6d632 Shared Rabbit Connection: SimpleConnection@47918f0b [delegate=amqp://guest@127.0.0.1:5672/, localPort= 63102], acknowledgeMode=AUTO local queue size=0
2018-10-04 13:30:34.518  INFO 98685 --- [cTaskExecutor-2] o.s.a.r.c.CachingConnectionFactory       : Attempting to connect to: [localhost:5672]
2018-10-04 13:30:39.559  INFO 98685 --- [cTaskExecutor-3] o.s.a.r.c.CachingConnectionFactory       : Created new connection: rabbitConnectionFactory#545b995e:3/SimpleConnection@39729070 [delegate=amqp://guest@127.0.0.1:5672/, localPort= 63123]

i.e. the container detects the closed connection and reestablishes it.

Whereas in your case, we see

WARN  | 2018-09-30 11:28:37.848 | [AMQP Connection 54.86.215.255:5672] |  |  | impl.ForgivingExceptionHandler (ForgivingExceptionHandler.java:120) - An unexpected connection driver error occured (Exception message: Connection reset)
ERROR | 2018-09-30 11:28:37.849 | [AMQP Connection 54.86.215.255:5672] |  |  | connection.CachingConnectionFactory$DefaultChannelCloseLogger (CachingConnectionFactory.java:1278) - Channel shutdown: connection error
ERROR | 2018-09-30 11:28:37.849 | [AMQP Connection 54.86.215.255:5672] |  |  | connection.CachingConnectionFactory$DefaultChannelCloseLogger (CachingConnectionFactory.java:1278) - Channel shutdown: connection error
INFO  | 2018-09-30 11:28:39.401 | [qtp1418129042-20] | ef04644f26045703 | ef04644f26045703 | connection.AbstractConnectionFactory (AbstractConnectionFactory.java:360) - Created new connection: connectionFactory#623d41c5:2/SimpleConnection@1beb3736 [delegate=amqp://******@54.86.215.255:5672/prod, localPort= 57360]

I see you are using a version that's over a year old; the latest 1.7.x version is 1.7.10; I don't recall any changes in this area, but can you upgrade?

What are these qtp1418129042-20 threads?

[garyrussell]

I just tested with 17.4 and I still see the (current) behavior - even with a prefetch, processing stops immediately and the consumer is restarted.

I have no idea what's happening. The sequence is this...

When the channel is shut down, the amqp-client calls the consumer's handleShutdownSignal(), in there we set the BlockingQueueConmsumers shutdown field with the exception. Then, each time we look for a message (even if prefetched) we check that field (and throw it). This is the signal to the container that it needs to go into recovery mode and restart the consumers.

I have no idea why, but the fact you still process prefetched records after the closure means the amqp-client never called the consumer
s handleShutdownSignal().

This code has been in place since early 2011 and I have never heard of such a problem before.

[slevinson]

we are actually using org.springframework.boot:spring-boot-starter-amqp:2.0.4.RELEASE
isn't it using spring-ampq of the same version?

in other services we are using org.springframework.boot:spring-boot-starter-amqp:1.5.9.RELEASE
and haven't experienced this problem, of course it is also different code.

[garyrussell]

Well, in the second log you sent, I see

WARN  | 2018-10-04 08:54:02.201 | [qtp1136705850-20] | 99e76713ff0c4270 | 99e76713ff0c4270 | health.AbstractHealthIndicator (AbstractHealthIndicator.java:46) - Health check failed
org.springframework.amqp.AmqpIOException: java.net.SocketException: Connection reset
	at org.springframework.amqp.rabbit.support.RabbitExceptionTranslator.convertRabbitAccessException(RabbitExceptionTranslator.java:71) ~[spring-rabbit-1.7.4.RELEASE.jar:?]
	at org.springframework.amqp.rabbit.connection.SimpleConnection.createChannel(SimpleConnection.java:63) ~[spring-rabbit-1.7.4.RELEASE.jar:?]

Hence my thought that this was what you are using.

Boot 2.0.4 will use spring-rabbit 2.0.5.RELEASE (unless overridden).
Boot 2.0.5 uses 2.0.6 - the release cycles are not aligned.

As I said, this code has been present for over 7 years without any problems being reported.

I am still suspicious about these qtp1418129042-20 threads and what they are doing - is it possible you have 2 connections? Even so, it's still not clear how this could possibly happen if heartbeats are present.