From 6a86f5231d300a61ce9e82f32a817261d714e3b6 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 22 Jan 2018 17:19:32 -0500 Subject: [PATCH] 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 --- .../udp/MulticastSendingMessageHandler.java | 68 +++++++++---------- .../ip/udp/UnicastSendingMessageHandler.java | 20 +++--- ...ramPacketMulticastSendingHandlerTests.java | 12 ++-- .../ip/udp/UdpChannelAdapterTests.java | 3 + 4 files changed, 52 insertions(+), 51 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java index b07633d4c5..bb8c6f38cb 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java @@ -100,56 +100,51 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler } @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); - } - if (this.timeToLive >= 0) { - socket.setTimeToLive(this.timeToLive); - } - setSocketAttributes(socket); - if (this.localAddress != null) { InetAddress whichNic = InetAddress.getByName(this.localAddress); - socket.setInterface(whichNic); + socket = new MulticastSocket(new InetSocketAddress(whichNic, ackPort)); } - this.multicastSocket = socket; + if (getSoReceiveBufferSize() > 0) { + socket.setReceiveBufferSize(this.getSoReceiveBufferSize()); + } + if (logger.isDebugEnabled()) { + logger.debug("Listening for acks on port: " + socket.getLocalPort()); + } + 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 class MulticastSendingMessageHandler extends UnicastSendingMessageHandler /** * Set the underlying {@link MulticastSocket} time to live property. - * * @param timeToLive {@link MulticastSocket#setTimeToLive(int)} */ public void setTimeToLive(int timeToLive) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java index c3a45ed70e..d467cfe04f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java +++ b/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 ackControl = Collections - .synchronizedMap(new HashMap()); + private volatile Map ackControl = + Collections.synchronizedMap(new HashMap()); private volatile int soReceiveBufferSize = -1; @@ -282,11 +282,7 @@ public class UnicastSendingMessageHandler extends 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 class UnicastSendingMessageHandler extends } 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 class UnicastSendingMessageHandler extends private void closeSocketIfNeeded() { if (this.socket != null) { - this.socket.close(); + try { + this.socket.close(); + } + catch (Exception e) { + } this.socket = null; } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java index c638da589b..7feb872689 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java +++ b/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.InetAddress; 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 @@ import org.springframework.messaging.Message; /** * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 */ public class DatagramPacketMulticastSendingHandlerTests { @@ -100,7 +102,7 @@ public class DatagramPacketMulticastSendingHandlerTests { } } }; - 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 class DatagramPacketMulticastSendingHandlerTests { assertTrue(received.await(10000, TimeUnit.MILLISECONDS)); handler.stop(); socket.close(); + executor.shutdownNow(); } @Test @@ -177,7 +180,7 @@ public class DatagramPacketMulticastSendingHandlerTests { } } }; - 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 class DatagramPacketMulticastSendingHandlerTests { assertTrue(ackSent.await(10000, TimeUnit.MILLISECONDS)); handler.stop(); socket.close(); + executor.shutdownNow(); } public void waitAckListening(UnicastSendingMessageHandler handler) throws InterruptedException { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java index eeeea7f0e5..e673bdb4e3 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java @@ -299,6 +299,7 @@ public class UdpChannelAdapterTests { assertNotNull(receivedMessage); assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload())); adapter.stop(); + handler.stop(); } @Test @@ -309,6 +310,8 @@ public class UdpChannelAdapterTests { // 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);