From 01b0e1c79ae96dba05ad343faf980c781da178ba 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** --- .../udp/MulticastSendingMessageHandler.java | 12 ++++------- .../ip/udp/UnicastSendingMessageHandler.java | 21 +++++++++---------- .../ip/udp/UdpChannelAdapterTests.java | 3 +++ 3 files changed, 17 insertions(+), 19 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 36c92f137a..18f0ea0cc0 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 @@ -125,13 +125,11 @@ 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 { @@ -174,7 +172,6 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler /** * 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) { @@ -183,7 +180,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 8abf33c6b8..01f171cd11 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,12 +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 { @@ -512,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 { @@ -530,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/UdpChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java index 22ab078fb8..fbe3baaa3c 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 @@ -294,6 +294,7 @@ public class UdpChannelAdapterTests { assertNotNull(receivedMessage); assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload())); adapter.stop(); + handler.stop(); } @Test @@ -302,6 +303,8 @@ public class UdpChannelAdapterTests { UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(0); 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);