INT-4366: Fix MulticastSendingMH race condition

JIRA: https://jira.spring.io/browse/INT-4366

The `MulticastSendingMessageHandler.getSocket()` doesn't guard around
`this.multicastSocket` property causing `NPE` and other inconsistency
in the multi-threaded environment

* Make the whole `MulticastSendingMessageHandler.getSocket()` as
`synchronized` like it is with the super method
* Reuse `closeSocketIfNeeded()` in the
`UnicastSendingMessageHandler.handleMessageInternal()`
* Fix type in the `UnicastSendingMessageHandler` logging message
* Fix `UdpChannelAdapterTests` for missed `BeanFactory` for the SpEL
and also `MulticastSendingMessageHandler.stop()` in one missed places

**Cherry-pick to 4.3.x**

# Conflicts:
#	spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java

Author: artembilan

spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java

@@ -100,56 +100,51 @@ public MulticastSendingMessageHandler(String address, int port,
 	}
 
 	@Override
-	protected DatagramSocket getSocket() throws IOException {
-		if (this.getTheSocket() == null) {
-			synchronized (this) {
-				createSocket();
-			}
+	protected synchronized DatagramSocket getSocket() throws IOException {
+		if (getTheSocket() == null) {
+			createSocket();
 		}
-		return this.getTheSocket();
+		return getTheSocket();
 	}
 
 	private void createSocket() throws IOException {
-		if (this.getTheSocket() == null) {
-			MulticastSocket socket;
-			if (this.isAcknowledge()) {
-				int ackPort = this.getAckPort();
-				if (this.localAddress == null) {
-					socket = ackPort == 0 ? new MulticastSocket() : new MulticastSocket(ackPort);
-				}
-				else {
-					InetAddress whichNic = InetAddress.getByName(this.localAddress);
-					socket = new MulticastSocket(new InetSocketAddress(whichNic, ackPort));
-				}
-				if (getSoReceiveBufferSize() > 0) {
-					socket.setReceiveBufferSize(this.getSoReceiveBufferSize());
-				}
-				if (logger.isDebugEnabled()) {
-					logger.debug("Listening for acks on port: " + socket.getLocalPort());
-				}
-				setSocket(socket);
-				updateAckAddress();
+		MulticastSocket socket;
+		if (this.isAcknowledge()) {
+			int ackPort = this.getAckPort();
+			if (this.localAddress == null) {
+				socket = ackPort == 0 ? new MulticastSocket() : new MulticastSocket(ackPort);
 			}
 			else {
-				socket = new MulticastSocket();
-				setSocket(socket);
+				InetAddress whichNic = InetAddress.getByName(this.localAddress);
+				socket = new MulticastSocket(new InetSocketAddress(whichNic, ackPort));
 			}
-			if (this.timeToLive >= 0) {
-				socket.setTimeToLive(this.timeToLive);
+			if (getSoReceiveBufferSize() > 0) {
+				socket.setReceiveBufferSize(this.getSoReceiveBufferSize());
 			}
-			setSocketAttributes(socket);
-			if (this.localAddress != null) {
-				InetAddress whichNic = InetAddress.getByName(this.localAddress);
-				socket.setInterface(whichNic);
+			if (logger.isDebugEnabled()) {
+				logger.debug("Listening for acks on port: " + socket.getLocalPort());
 			}
-			this.multicastSocket = socket;
+			setSocket(socket);
+			updateAckAddress();
+		}
+		else {
+			socket = new MulticastSocket();
+			setSocket(socket);
+		}
+		if (this.timeToLive >= 0) {
+			socket.setTimeToLive(this.timeToLive);
+		}
+		setSocketAttributes(socket);
+		this.multicastSocket = socket;
+		if (this.localAddress != null) {
+			InetAddress whichNic = InetAddress.getByName(this.localAddress);
+			socket.setInterface(whichNic);
 		}
 	}
 
 
 	/**
 	 * If acknowledge = true; how many acks needed for success.
-	 *
 	 * @param minAcksForSuccess The minimum number of acks that will represent success.
 	 */
 	public void setMinAcksForSuccess(int minAcksForSuccess) {
@@ -158,7 +153,6 @@ public void setMinAcksForSuccess(int minAcksForSuccess) {
 
 	/**
 	 * Set the underlying {@link MulticastSocket} time to live property.
-	 *
 	 * @param timeToLive {@link MulticastSocket#setTimeToLive(int)}
 	 */
 	public void setTimeToLive(int timeToLive) {

spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2001-2017 the original author or authors.
+ * Copyright 2001-2018 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.
@@ -83,8 +83,8 @@ public class UnicastSendingMessageHandler extends
 
 	private volatile int ackCounter = 1;
 
-	private volatile Map<String, CountDownLatch> ackControl = Collections
-			.synchronizedMap(new HashMap<String, CountDownLatch>());
+	private volatile Map<String, CountDownLatch> ackControl =
+			Collections.synchronizedMap(new HashMap<String, CountDownLatch>());
 
 	private volatile int soReceiveBufferSize = -1;
 
@@ -282,11 +282,7 @@ public void handleMessageInternal(Message<?> message) throws MessageHandlingExce
 			throw e;
 		}
 		catch (Exception e) {
-			try {
-				this.socket.close();
-			}
-			catch (Exception e1) { }
-			this.socket = null;
+			closeSocketIfNeeded();
 			throw new MessageHandlingException(message, "failed to send UDP packet", e);
 		}
 		finally {
@@ -511,7 +507,7 @@ public void run() {
 		}
 		catch (IOException e) {
 			if (this.socket != null && !this.socket.isClosed()) {
-				logger.error("Error on UDP Acknowledge thread:" + e.getMessage());
+				logger.error("Error on UDP Acknowledge thread: " + e.getMessage());
 			}
 		}
 		finally {
@@ -529,7 +525,11 @@ public void restartAckThread() {
 
 	private void closeSocketIfNeeded() {
 		if (this.socket != null) {
-			this.socket.close();
+			try {
+				this.socket.close();
+			}
+			catch (Exception e) {
+			}
 			this.socket = null;
 		}
 	}

spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -26,7 +26,7 @@
 import java.net.InetSocketAddress;
 import java.net.MulticastSocket;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -45,6 +45,8 @@
 /**
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class DatagramPacketMulticastSendingHandlerTests {
@@ -100,7 +102,7 @@ public void run() {
 				}
 			}
 		};
-		Executor executor = Executors.newFixedThreadPool(2);
+		ExecutorService executor = Executors.newFixedThreadPool(2);
 		executor.execute(catcher);
 		executor.execute(catcher);
 		assertTrue(listening.await(10000, TimeUnit.MILLISECONDS));
@@ -112,6 +114,7 @@ public void run() {
 		assertTrue(received.await(10000, TimeUnit.MILLISECONDS));
 		handler.stop();
 		socket.close();
+		executor.shutdownNow();
 	}
 
 	@Test
@@ -177,7 +180,7 @@ public void run() {
 				}
 			}
 		};
-		Executor executor = Executors.newFixedThreadPool(2);
+		ExecutorService executor = Executors.newFixedThreadPool(2);
 		executor.execute(catcher);
 		executor.execute(catcher);
 		assertTrue(listening.await(10000, TimeUnit.MILLISECONDS));
@@ -195,6 +198,7 @@ public void run() {
 		assertTrue(ackSent.await(10000, TimeUnit.MILLISECONDS));
 		handler.stop();
 		socket.close();
+		executor.shutdownNow();
 	}
 
 	public void waitAckListening(UnicastSendingMessageHandler handler) throws InterruptedException {

spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java

@@ -299,6 +299,7 @@ public void testMulticastSender() throws Exception {
 		assertNotNull(receivedMessage);
 		assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload()));
 		adapter.stop();
+		handler.stop();
 	}
 
 	@Test
@@ -309,6 +310,8 @@ public void testUnicastReceiverException() throws Exception {
 //		SocketUtils.setLocalNicIfPossible(adapter);
 		adapter.setOutputChannel(channel);
 		ServiceActivatingHandler handler = new ServiceActivatingHandler(new FailingService());
+		handler.setBeanFactory(mock(BeanFactory.class));
+		handler.afterPropertiesSet();
 		channel.subscribe(handler);
 		QueueChannel errorChannel = new QueueChannel();
 		adapter.setErrorChannel(errorChannel);