vendredi 8 mai 2015

spring-amqp with RabbitMQ does not shutdown properly

When consumer not finishes execution within SimpleMessageListenerContainer shutdownTimeout after AnnotationConfigApplicationContext.close() invocation, my Spring application hangs.

Listener:

public class LongRunningMessageListener implements MessageListener {

    @Override
    public void onMessage(Message message) {
        System.out.println("Got " + new String(message.getBody()));
        try {
            Thread.sleep(5000L);
        } catch (InterruptedException e) {
            System.out.println("Interrupted");
            Thread.currentThread().interrupt();
        }
        System.out.println("Finished execution");
    }
}

Configuration:

@Configuration
public class Config {

    @Bean
    public MessageListenerContainer messageListenerContainer() {
        SimpleMessageListenerContainer messageListenerContainer = new SimpleMessageListenerContainer(connectionFactory());
        messageListenerContainer.setQueueNames("myqueue");
        messageListenerContainer.setMessageListener(new LongRunningMessageListener());
        messageListenerContainer.setShutdownTimeout(1000);
        return messageListenerContainer;
    }

    @Bean
    public ConnectionFactory connectionFactory() {
        return new CachingConnectionFactory("localhost");
    }
}

Main:

public static void main(String[] args) throws InterruptedException {
    AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext(Config.class);
    Thread.sleep(1000L);
    applicationContext.close();
}

Output:

мая 08, 2015 3:43:21 PM org.springframework.context.annotation.AnnotationConfigApplicationContext prepareRefresh
INFO: Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@1a86f2f1: startup date [Fri May 08 15:43:21 MSK 2015]; root of context hierarchy
мая 08, 2015 3:43:22 PM org.springframework.context.support.DefaultLifecycleProcessor start
INFO: Starting beans in phase 2147483647
мая 08, 2015 3:43:22 PM org.springframework.amqp.rabbit.connection.CachingConnectionFactory createBareConnection
INFO: Created new connection: SimpleConnection@4025f2f5 [delegate=amqp://guest@127.0.0.1:5672/]
Got foo1
мая 08, 2015 3:43:23 PM org.springframework.context.annotation.AnnotationConfigApplicationContext doClose
INFO: Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@1a86f2f1: startup date [Fri May 08 15:43:21 MSK 2015]; root of context hierarchy
мая 08, 2015 3:43:23 PM org.springframework.context.support.DefaultLifecycleProcessor stop
INFO: Stopping beans in phase 2147483647
мая 08, 2015 3:43:23 PM org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer doShutdown
INFO: Waiting for workers to finish.
мая 08, 2015 3:43:24 PM org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer doShutdown
INFO: Workers not finished.  Forcing connections to close.
Finished execution
мая 08, 2015 3:43:27 PM org.springframework.amqp.rabbit.connection.CachingConnectionFactory createBareConnection
INFO: Created new connection: SimpleConnection@260da64e [delegate=amqp://guest@127.0.0.1:5672/]
мая 08, 2015 3:43:27 PM org.springframework.amqp.rabbit.connection.CachingConnectionFactory shutdownCompleted
SEVERE: Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - unknown delivery tag 1, class-id=60, method-id=80)

After that application is working infinitely.

Working threads:

Live threads: 14
Daemon threads: 10

Non daemon threads:

"pool-1-thread-10" #22 prio=5 os_prio=31 tid=0x00007fd48c93c800 nid=0x5a07 waiting on condition [0x0000000129eb3000]
   java.lang.Thread.State: WAITING (parking)
    at sun.misc.Unsafe.park(Native Method)
    - parking to wait for  <0x0000000796333168> (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:1067)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

   Locked ownable synchronizers:
    - None

"pool-1-thread-9" #21 prio=5 os_prio=31 tid=0x00007fd48b9d8000 nid=0x540b waiting on condition [0x0000000129db0000]
   java.lang.Thread.State: TIMED_WAITING (parking)
    at sun.misc.Unsafe.park(Native Method)
    - parking to wait for  <0x000000079633dfb0> (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:1067)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

   Locked ownable synchronizers:
    - None

"AMQP Connection 127.0.0.1:5672" #20 prio=5 os_prio=31 tid=0x00007fd48d186000 nid=0x530b runnable [0x0000000129cad000]
   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:170)
    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 <0x000000079632e8f0> (a java.io.BufferedInputStream)
    at java.io.DataInputStream.readUnsignedByte(DataInputStream.java:288)
    at com.rabbitmq.client.impl.Frame.readFrom(Frame.java:95)
    at com.rabbitmq.client.impl.SocketFrameHandler.readFrame(SocketFrameHandler.java:139)
    - locked <0x000000079632e8d0> (a java.io.DataInputStream)
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:534)
    at java.lang.Thread.run(Thread.java:745)

   Locked ownable synchronizers:
    - None

"DestroyJavaVM" #19 prio=5 os_prio=31 tid=0x00007fd48d002000 nid=0x1303 waiting on condition [0x0000000000000000]
   java.lang.Thread.State: RUNNABLE

   Locked ownable synchronizers:
    - None

Everything is fine if I increase shutdownTimeout, but timeout behavior looks like a bug.

Spring versions:

<dependencies>
    <dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-amqp</artifactId>
        <version>1.4.5.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-rabbit</artifactId>
        <version>1.4.5.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-context</artifactId>
        <version>4.1.5.RELEASE</version>
    </dependency>
</dependencies>

Aucun commentaire:

Enregistrer un commentaire