DATAREDIS-415 - Handle InterruptedException correctly in RedisMessageListenerContainer.

Set interrupted bit when catching InterruptedException. Exit loops when catching InterruptedException to free resources.

Original Pull Request: #168

src/main/java/org/springframework/data/redis/connection/ClusterCommandExecutor.java

@@ -242,8 +242,11 @@ public class ClusterCommandExecutor implements DisposableBean {
 							exceptions.put(entry.getKey(), ex != null ? ex : e.getCause());
 						} catch (InterruptedException e) {
 
+							Thread.currentThread().interrupt();
+
 							RuntimeException ex = convertToDataAccessExeption((Exception) e.getCause());
 							exceptions.put(entry.getKey(), ex != null ? ex : e.getCause());
+							break;
 						}
 					}
 				}

src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java

@@ -67,6 +67,7 @@ import org.springframework.util.ErrorHandler;
  * @author Jennifer Hickey
  * @author Way Joke
  * @author Thomas Darimont
+ * @author Mark Paluch
  */
 public class RedisMessageListenerContainer implements InitializingBean, DisposableBean, BeanNameAware, SmartLifecycle {
 
@@ -204,6 +205,9 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 						monitor.wait(initWait);
 					} catch (InterruptedException e) {
 						// stop waiting
+						Thread.currentThread().interrupt();
+						running = false;
+						return;
 					}
 				}
 			}
@@ -658,6 +662,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 				Thread.sleep(this.recoveryInterval);
 			} catch (InterruptedException interEx) {
 				logger.debug("Thread interrupted while sleeping the recovery interval");
+				Thread.currentThread().interrupt();
 			}
 		}
 	}
@@ -700,7 +705,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 								try {
 									Thread.sleep(WAIT);
 								} catch (InterruptedException ex) {
-									done = true;
+									Thread.currentThread().interrupt();
+									return;
 								}
 							}
 						}
@@ -730,7 +736,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 
 				boolean asyncConnection = ConnectionUtils.isAsync(connectionFactory);
 
-				// NB: async drivers' Xsubscribe calls block, so we notify the RDMLC before performing the actual subscription.
+				// NB: sync drivers' Xsubscribe calls block, so we notify the RDMLC before performing the actual subscription.
 				if (!asyncConnection) {
 					synchronized (monitor) {
 						monitor.notify();
@@ -849,6 +855,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 								localMonitor.wait(subscriptionWait);
 							} catch (InterruptedException e) {
 								// Stop waiting
+								Thread.currentThread().interrupt();
 							}
 						}
 						if (!subscriptionTaskRunning) {
@@ -1004,16 +1011,18 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab
 
 			long startTime = System.currentTimeMillis();
 
-			while (!timedOut(startTime, timeout)) {
-				if (condition.passes()) {
-					return true;
-				}
-				try {
+			try {
+				while (!timedOut(startTime, timeout)) {
+					if (condition.passes()) {
+						return true;
+					}
+
 					Thread.sleep(100);
-				} catch (InterruptedException e) {
-					Thread.currentThread().interrupt();
 				}
+			} catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
 			}
+
 			return false;
 		}
 

src/test/java/org/springframework/data/redis/listener/RedisMessageListenerContainerTest.java

@@ -0,0 +1,108 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.redis.listener;
+
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doAnswer;
+
+import java.util.concurrent.Executor;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.mockito.stubbing.Answer;
+import org.springframework.core.task.SyncTaskExecutor;
+import org.springframework.data.redis.SettingsUtils;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+
+/**
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class RedisMessageListenerContainerTest {
+
+	private JedisConnectionFactory connectionFactory;
+	private RedisMessageListenerContainer container;
+
+	private final Object handler = new Object() {
+		@SuppressWarnings("unused")
+		public void handleMessage(Object message) {
+		}
+	};
+
+	private final MessageListenerAdapter adapter = new MessageListenerAdapter(handler);
+
+	@Mock private Executor executor;
+
+	@Before
+	public void before() throws Exception {
+
+		connectionFactory = new JedisConnectionFactory();
+		connectionFactory.setPort(SettingsUtils.getPort());
+		connectionFactory.setHostName(SettingsUtils.getHost());
+		connectionFactory.setDatabase(2);
+
+		connectionFactory.afterPropertiesSet();
+
+		container = new RedisMessageListenerContainer();
+		container.setConnectionFactory(connectionFactory);
+		container.setBeanName("container");
+		container.setTaskExecutor(new SyncTaskExecutor());
+		container.setSubscriptionExecutor(executor);
+		container.afterPropertiesSet();
+	}
+
+	/*
+	 * @see DATAREDIS-415
+	 */
+	@Test
+	public void interruptAtStart() throws Exception {
+
+		final Thread main = Thread.currentThread();
+
+		// interrupt thread once Executor.execute is called
+		doAnswer(new Answer() {
+			@Override
+			public Object answer(final InvocationOnMock invocationOnMock) throws Throwable {
+				main.interrupt();
+				return null;
+			}
+		}).when(executor).execute(any(Runnable.class));
+
+		container.addMessageListener(adapter, new ChannelTopic("a"));
+		container.start();
+
+		// reset the interrupted flag to not destroy the teardown
+		assertThat(Thread.interrupted(), is(true));
+
+		assertThat(container.isRunning(), is(false));
+	}
+
+	@After
+	public void tearDown() throws Exception {
+
+		container.destroy();
+		connectionFactory.destroy();
+	}
+}