INT-4366: Fix MulticastSendingMessageHandler

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

Fix race condition in the `MulticastSendingMessageHandler` around
`multicastSocket` and super `socket` properties.

* Synchronize around `this` and check for the `multicastSocket == null`.
This let the `MulticastSendingMessageHandler` to fully configure and
prepare the socket for use.
* Remove `socket.setInterface(whichNic)` since it is populated by the
`InetSocketAddress` ctor before

**Cherry-pick to 4.3.x**

Author: artembilan

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2001-2016 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.
@@ -35,6 +35,8 @@
  * determine success.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler {
@@ -101,49 +103,45 @@ public MulticastSendingMessageHandler(String address, int port,
 
 	@Override
 	protected DatagramSocket getSocket() throws IOException {
-		if (this.getTheSocket() == null) {
+		if (this.multicastSocket == null) {
 			synchronized (this) {
-				createSocket();
+				if (this.multicastSocket == 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 (isAcknowledge()) {
+			int ackPort = 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(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;
 	}
 
 
@@ -153,7 +151,7 @@ private void createSocket() throws IOException {
 	 * @param minAcksForSuccess The minimum number of acks that will represent success.
 	 */
 	public void setMinAcksForSuccess(int minAcksForSuccess) {
-		this.setAckCounter(minAcksForSuccess);
+		setAckCounter(minAcksForSuccess);
 	}
 
 	/**