GH-3526: Fix Infinite Loop in FailoverCConnFactory

Resolves #3526

`FailoverClientConnectionFactory`

The logic to detect we had iterated over all factories and including
the one from which the previous connection was established was incorrect,
causing an infite loop until one of the factory connections was successful.

Change the logic to detect we have reset the iterator and the current failure
is from the same factory as the one from which the previous connection was
established.

**cherry-pick to 5.4.x, 5.3.x**

* Add diagnostics.

* Fix race in test.

* More race fixes and diagnostics.

* Remove diagnostics.

Author: garyrussell

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -264,7 +264,7 @@ private synchronized void findAConnection() throws InterruptedException {
 			if (!this.factoryIterator.hasNext()) {
 				this.factoryIterator = this.connectionFactories.iterator();
 			}
-			boolean retried = false;
+			boolean restartedList = false;
 			while (!success) {
 				try {
 					nextFactory = this.factoryIterator.next();
@@ -282,17 +282,18 @@ private synchronized void findAConnection() throws InterruptedException {
 								+ e.toString()
 								+ ", trying another");
 					}
+					if (restartedList && (lastFactoryToTry == null || lastFactoryToTry.equals(nextFactory))) {
+						logger.debug("Failover failed to find a connection");
+						/*
+						 *  We've tried every factory including the
+						 *  one the current connection was on.
+						 */
+						this.open = false;
+						throw e;
+					}
 					if (!this.factoryIterator.hasNext()) {
-						if (retried && (lastFactoryToTry == null || lastFactoryToTry.equals(nextFactory))) {
-							/*
-							 *  We've tried every factory including the
-							 *  one the current connection was on.
-							 */
-							this.open = false;
-							throw e;
-						}
 						this.factoryIterator = this.connectionFactories.iterator();
-						retried = true;
+						restartedList = true;
 					}
 				}
 			}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java

@@ -316,10 +316,14 @@ public TcpListener getListener() {
 			return this.testListener;
 		}
 		if (this.manualListenerRegistration && !this.testFailed) {
-			if (this.logger.isDebugEnabled()) {
+			boolean debugEnabled = this.logger.isDebugEnabled();
+			if (debugEnabled) {
 				this.logger.debug(getConnectionId() + " Waiting for listener registration");
 			}
 			waitForListenerRegistration();
+			if (debugEnabled) {
+				this.logger.debug(getConnectionId() + " Listener registered");
+			}
 		}
 		return this.listener;
 	}

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactoryTests.java

@@ -28,8 +28,11 @@
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.when;
 
+import java.io.BufferedReader;
 import java.io.IOException;
+import java.io.InputStreamReader;
 import java.io.UncheckedIOException;
+import java.net.ServerSocket;
 import java.net.Socket;
 import java.nio.channels.SocketChannel;
 import java.util.ArrayList;
@@ -41,7 +44,9 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
-import org.junit.Test;
+import javax.net.ServerSocketFactory;
+
+import org.junit.jupiter.api.Test;
 import org.mockito.InOrder;
 import org.mockito.Mockito;
 
@@ -62,6 +67,7 @@
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 /**
  * @author Gary Russell
@@ -173,7 +179,7 @@ private void testRefreshShared(boolean closeOnRefresh, long interval) throws Exc
 		inOrder.verifyNoMoreInteractions();
 	}
 
-	@Test(expected = UncheckedIOException.class)
+	@Test
 	public void testFailoverAllDead() throws Exception {
 		AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
 		AbstractClientConnectionFactory factory2 = mock(AbstractClientConnectionFactory.class);
@@ -193,10 +199,47 @@ public void testFailoverAllDead() throws Exception {
 		FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
 		failoverFactory.start();
 		GenericMessage<String> message = new GenericMessage<String>("foo");
-		failoverFactory.getConnection().send(message);
+		assertThatExceptionOfType(UncheckedIOException.class).isThrownBy(() ->
+			failoverFactory.getConnection().send(message));
 		Mockito.verify(conn2).send(message);
 	}
 
+	@Test
+	void failoverAllDeadAfterSuccess() throws Exception {
+		ServerSocket ss1 = ServerSocketFactory.getDefault().createServerSocket(0);
+		ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor();
+		exec.initialize();
+		exec.submit(() -> {
+			Socket accepted = ss1.accept();
+			BufferedReader br = new BufferedReader(new InputStreamReader(accepted.getInputStream()));
+			br.readLine();
+			accepted.getOutputStream().write("ok\r\n".getBytes());
+			accepted.close();
+			ss1.close();
+			return true;
+		});
+		TcpNetClientConnectionFactory cf1 = new TcpNetClientConnectionFactory("localhost", ss1.getLocalPort());
+		AbstractClientConnectionFactory cf2 = mock(AbstractClientConnectionFactory.class);
+		doThrow(new UncheckedIOException(new IOException("fail"))).when(cf2).getConnection();
+		CountDownLatch latch = new CountDownLatch(2);
+		cf1.setApplicationEventPublisher(event -> {
+			if (event instanceof TcpConnectionCloseEvent) {
+				latch.countDown();
+			}
+		});
+		cf2.setApplicationEventPublisher(event -> { });
+		FailoverClientConnectionFactory fccf = new FailoverClientConnectionFactory(List.of(cf1, cf2));
+		fccf.registerListener(msf -> {
+			latch.countDown();
+			return false;
+		});
+		fccf.start();
+		fccf.getConnection().send(new GenericMessage<>("test"));
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThatExceptionOfType(UncheckedIOException.class).isThrownBy(() ->
+				fccf.getConnection().send(new GenericMessage<>("test")));
+	}
+
 	@Test
 	public void testFailoverAllDeadButOriginalOkAgain() throws Exception {
 		AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
@@ -228,7 +271,7 @@ public void testFailoverAllDeadButOriginalOkAgain() throws Exception {
 		Mockito.verify(conn1, times(2)).send(message);
 	}
 
-	@Test(expected = UncheckedIOException.class)
+	@Test
 	public void testFailoverConnectNone() throws Exception {
 		AbstractClientConnectionFactory factory1 = mock(AbstractClientConnectionFactory.class);
 		AbstractClientConnectionFactory factory2 = mock(AbstractClientConnectionFactory.class);
@@ -242,7 +285,8 @@ public void testFailoverConnectNone() throws Exception {
 		FailoverClientConnectionFactory failoverFactory = new FailoverClientConnectionFactory(factories);
 		failoverFactory.start();
 		GenericMessage<String> message = new GenericMessage<String>("foo");
-		failoverFactory.getConnection().send(message);
+		assertThatExceptionOfType(UncheckedIOException.class).isThrownBy(() ->
+			failoverFactory.getConnection().send(message));
 	}
 
 	@Test