Fix JDBC tests and some upgrades

https://build.spring.io/browse/INT-FATS5IC-430

Looks like there is a race condition when our polling rate around
data base is too fast and we have access to the DB files even during
application context close.

* Stop polling channel adapter in the tests explicitly after test methods
* Increase some tests performance decreasing timeouts to wait
* Upgrade to Reactor-3.1.5, AssertJ-3.9.1, Derby-10.14.1.0 and some
Gradle plugins

* Upgrade to `reactor-netty-0.7.5`
* Remove workaround from the `StompServerIntegrationTests`

* Upgrade to Spring Data Kay SR5

* Upgrade to Spring Security 5.0.3
* Increase timeouts and performance in the `StreamTransformerParserTests`

build.gradle

@@ -3,7 +3,7 @@ buildscript {
 		maven { url 'https://repo.spring.io/plugins-release' }
 	}
 	dependencies {
-		classpath 'io.spring.gradle:dependency-management-plugin:1.0.3.RELEASE'
+		classpath 'io.spring.gradle:dependency-management-plugin:1.0.4.RELEASE'
 		classpath 'io.spring.gradle:spring-io-plugin:0.0.8.RELEASE'
 		classpath 'io.spring.gradle:docbook-reference-plugin:0.3.1'
 		classpath 'org.asciidoctor:asciidoctor-gradle-plugin:1.5.0'
@@ -11,7 +11,7 @@ buildscript {
 }
 
 plugins {
-	id 'org.sonarqube' version '2.5'
+	id 'org.sonarqube' version '2.6.1'
 }
 
 description = 'Spring Integration'
@@ -90,13 +90,13 @@ subprojects { subproject ->
 		activeMqVersion = '5.15.2'
 		apacheSshdVersion = '1.6.0'
 		aspectjVersion = '1.8.13'
-		assertjVersion = '2.6.0'
+		assertjVersion = '3.9.1'
 		boonVersion = '0.34'
 		commonsDbcp2Version = '2.1.1'
 		commonsIoVersion = '2.4'
 		commonsNetVersion = '3.5'
 		curatorVersion = '2.11.1'
-		derbyVersion = '10.13.1.1'
+		derbyVersion = '10.14.1.0'
 		eclipseLinkVersion = '2.6.4'
 		ftpServerVersion = '1.1.1'
 		groovyVersion = '2.4.12'
@@ -125,17 +125,17 @@ subprojects { subproject ->
 		mysqlVersion = '6.0.6'
 		pahoMqttClientVersion = '1.2.0'
 		postgresVersion = '42.0.0'
-		reactorNettyVersion = '0.7.4.RELEASE'
+		reactorNettyVersion = '0.7.5.RELEASE'
-		reactorVersion = '3.1.4.RELEASE'
+		reactorVersion = '3.1.5.RELEASE'
 		romeToolsVersion = '1.8.0'
 		servletApiVersion = '4.0.0'
 		smackVersion = '4.2.2'
 		springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.0.2.RELEASE'
-		springDataJpaVersion = '2.0.4.RELEASE'
+		springDataJpaVersion = '2.0.5.RELEASE'
-		springDataMongoVersion = '2.0.4.RELEASE'
+		springDataMongoVersion = '2.0.5.RELEASE'
-		springDataRedisVersion = '2.0.4.RELEASE'
+		springDataRedisVersion = '2.0.5.RELEASE'
-		springGemfireVersion = '2.0.4.RELEASE'
+		springGemfireVersion = '2.0.5.RELEASE'
-		springSecurityVersion = '5.0.2.RELEASE'
+		springSecurityVersion = '5.0.3.RELEASE'
 		springSocialTwitterVersion = '1.1.2.RELEASE'
 		springRetryVersion = '1.2.2.RELEASE'
 		springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.0.4.RELEASE'

spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests-context.xml

@@ -20,7 +20,7 @@
 	<stream-transformer input-channel="directInput" output-channel="output"/>
 
 	<stream-transformer input-channel="queueInput" output-channel="output">
-		<poller fixed-delay="10000"/>
+		<poller fixed-delay="1"/>
 	</stream-transformer>
 
 	<chain input-channel="charsetChannel" output-channel="output">

spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-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.
@@ -39,6 +39,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 /**
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
@@ -64,7 +65,7 @@ public class StreamTransformerParserTests {
 	@Test
 	public void directChannelWithStringMessage() {
 		this.directInput.send(new GenericMessage<InputStream>(new ByteArrayInputStream("foo".getBytes())));
-		Message<?> result = output.receive(0);
+		Message<?> result = output.receive(10000);
 		assertNotNull(result);
 		assertArrayEquals("foo".getBytes(), (byte[]) result.getPayload());
 	}
@@ -72,7 +73,7 @@ public class StreamTransformerParserTests {
 	@Test
 	public void queueChannelWithStringMessage() {
 		this.queueInput.send(new GenericMessage<InputStream>(new ByteArrayInputStream("foo".getBytes())));
-		Message<?> result = output.receive(3000);
+		Message<?> result = output.receive(10000);
 		assertNotNull(result);
 		assertArrayEquals("foo".getBytes(), (byte[]) result.getPayload());
 	}
@@ -80,7 +81,7 @@ public class StreamTransformerParserTests {
 	@Test
 	public void charset() {
 		this.charsetChannel.send(new GenericMessage<InputStream>(new ByteArrayInputStream("foo".getBytes())));
-		Message<?> result = output.receive(0);
+		Message<?> result = output.receive(10000);
 		assertNotNull(result);
 		assertEquals("foo", result.getPayload());
 	}

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/JdbcMessageStoreChannelIntegrationTests-context.xml

@@ -65,6 +65,7 @@
 	<bean id="transactionManager"
 		  class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
 		<property name="dataSource" ref="dataSource"/>
+		<property name="defaultTimeout" value="10"/>
 	</bean>
 
 	<int:header-enricher input-channel="routingSlip" output-channel="input">

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/JdbcMessageStoreChannelIntegrationTests.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.
@@ -31,6 +31,7 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import org.hamcrest.Matchers;
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -39,6 +40,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.core.serializer.support.SerializationFailedException;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.endpoint.AbstractEndpoint;
 import org.springframework.integration.store.MessageGroup;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageDeliveryException;
@@ -81,17 +83,28 @@ public class JdbcMessageStoreChannelIntegrationTests {
 	@Autowired
 	private MessageChannel routingSlip;
 
+	@Autowired
+	@Qualifier("service-activator")
+	private AbstractEndpoint serviceActivator;
+
 	@Before
 	public void clear() {
 		Service.reset(1);
 		for (MessageGroup group : messageStore) {
 			messageStore.removeMessageGroup(group.getGroupId());
 		}
+
+		this.serviceActivator.start();
+	}
+
+	@After
+	public void tearDown() {
+		this.serviceActivator.stop();
 	}
 
 	@Test
 	public void testSendAndActivate() throws Exception {
-		input.send(new GenericMessage<String>("foo"));
+		input.send(new GenericMessage<>("foo"));
 		Service.await(10000);
 		assertEquals(1, Service.messages.size());
 	}
@@ -99,7 +112,7 @@ public class JdbcMessageStoreChannelIntegrationTests {
 	@Test
 	public void testSendAndActivateWithRollback() throws Exception {
 		Service.fail = true;
-		input.send(new GenericMessage<String>("foo"));
+		input.send(new GenericMessage<>("foo"));
 		Service.await(10000);
 		assertThat(Service.messages.size(), Matchers.greaterThanOrEqualTo(1));
 		// After a rollback in the poller the message is still waiting to be delivered
@@ -126,10 +139,10 @@ public class JdbcMessageStoreChannelIntegrationTests {
 
 			synchronized (storeLock) {
 
-				boolean result1 = input.send(new GenericMessage<String>("foo"), 100L);
+				boolean result1 = input.send(new GenericMessage<>("foo"), 100L);
 				// This will time out because the transaction has not committed yet
 				try {
-					Service.await(300);
+					Service.await(100);
 					fail("Expected timeout");
 				}
 				catch (Exception e) {
@@ -174,7 +187,7 @@ public class JdbcMessageStoreChannelIntegrationTests {
 	}
 
 	@Test
-	public void testSameTransactionSendAndReceive() throws Exception {
+	public void testSameTransactionSendAndReceive() {
 
 		final StopWatch stopWatch = new StopWatch();
 		DefaultTransactionDefinition transactionDefinition = new DefaultTransactionDefinition();
@@ -225,7 +238,7 @@ public class JdbcMessageStoreChannelIntegrationTests {
 	@Test
 	public void testWithRoutingSlip() {
 		try {
-			this.routingSlip.send(new GenericMessage<String>("foo"));
+			this.routingSlip.send(new GenericMessage<>("foo"));
 			fail("MessageDeliveryException expected");
 		}
 		catch (Exception e) {
@@ -241,7 +254,7 @@ public class JdbcMessageStoreChannelIntegrationTests {
 
 		private static boolean fail = false;
 
-		private static List<String> messages = new CopyOnWriteArrayList<String>();
+		private static List<String> messages = new CopyOnWriteArrayList<>();
 
 		private static CountDownLatch latch;
 

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/JdbcMessageStoreChannelOnePollerIntegrationTests-context.xml

@@ -59,6 +59,7 @@
 	<bean id="transactionManager"
 		class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
 		<property name="dataSource" ref="dataSource" />
+		<property name="defaultTimeout" value="10"/>
 	</bean>
 
 </beans>

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/JdbcMessageStoreChannelOnePollerIntegrationTests.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,6 +26,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -33,6 +34,7 @@ import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.endpoint.AbstractEndpoint;
 import org.springframework.integration.store.MessageGroup;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
@@ -42,6 +44,11 @@ import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.util.StopWatch;
 
+/**
+ * @author Dave Syer
+ * @author Gary Russell
+ * @author Artem Bilan
+ */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
 @DirtiesContext // close at the end after class
@@ -63,50 +70,59 @@ public class JdbcMessageStoreChannelOnePollerIntegrationTests {
 	@Autowired
 	private PlatformTransactionManager transactionManager;
 
+	@Autowired
+	@Qualifier("service-relay")
+	private AbstractEndpoint serviceRelay;
+
 	@Before
 	public void clear() {
-		for (MessageGroup group : messageStore) {
+		for (MessageGroup group : this.messageStore) {
-			messageStore.removeMessageGroup(group.getGroupId());
+			this.messageStore.removeMessageGroup(group.getGroupId());
 		}
 	}
 
-	@Test
+	@After
-	// @Repeat(50)
+	public void tearDown() {
-	public void testSameTransactionDifferentChannelSendAndReceive() throws Exception {
+		this.serviceRelay.stop();
+	}
 
+	@Test
+	public void testSameTransactionDifferentChannelSendAndReceive() throws Exception {
 		Service.reset(1);
-		assertNull(durable.receive(100L));
+		assertNull(this.durable.receive(100L));
-		assertNull(relay.receive(100L));
+		assertNull(this.relay.receive(100L));
 		final StopWatch stopWatch = new StopWatch();
 
-		boolean result = new TransactionTemplate(transactionManager).execute(status -> {
+		boolean result =
+				new TransactionTemplate(this.transactionManager)
+						.execute(status -> {
 
-			synchronized (storeLock) {
+							synchronized (this.storeLock) {
 
-				boolean result1 = relay.send(new GenericMessage<String>("foo"), 500L);
+								boolean result1 = this.relay.send(new GenericMessage<>("foo"), 500L);
-				// This will time out because the transaction has not committed yet
+								// This will time out because the transaction has not committed yet
-				try {
+								try {
-					Service.await(10000);
+									Service.await(100);
-					fail("Expected timeout");
+									fail("Expected timeout");
-				}
+								}
-				catch (Exception e) {
+								catch (Exception e) {
-					// expected
+									// expected
-				}
+								}
 
-				try {
+								try {
-					stopWatch.start();
+									stopWatch.start();
-					// It hasn't arrive yet because we are still in the sending transaction
+									// It hasn't arrive yet because we are still in the sending transaction
-					assertNull(durable.receive(100L));
+									assertNull(this.durable.receive(100L));
-				}
+								}
-				finally {
+								finally {
-					stopWatch.stop();
+									stopWatch.stop();
-				}
+								}
 
-				return result1;
+								return result1;
 
-			}
+							}
 
-		});
+						});
 
 		assertTrue("Could not send message", result);
 		// If the poll blocks in the RDBMS there is no way for the queue to respect the timeout
@@ -124,20 +140,21 @@ public class JdbcMessageStoreChannelOnePollerIntegrationTests {
 		 *
 		 * With the storeLock: It doesn't deadlock as long as the lock is injected into the poller as well.
 		 */
-		new TransactionTemplate(transactionManager).execute(status -> {
+		new TransactionTemplate(this.transactionManager)
-			synchronized (storeLock) {
+				.execute(status -> {
+					synchronized (this.storeLock) {
 
-				try {
+						try {
-					stopWatch.start();
+							stopWatch.start();
-					durable.receive(100L);
+							this.durable.receive(100L);
-					return null;
+							return null;
-				}
+						}
-				finally {
+						finally {
-					stopWatch.stop();
+							stopWatch.stop();
-				}
+						}
 
-			}
+					}
-		});
+				});
 
 		// If the poll blocks in the RDBMS there is no way for the queue to respect the timeout
 		assertTrue("Timed out waiting for receive", stopWatch.getTotalTimeMillis() < 10000);
@@ -145,6 +162,7 @@ public class JdbcMessageStoreChannelOnePollerIntegrationTests {
 	}
 
 	public static class Service {
+
 		private static boolean fail = false;
 
 		private static List<String> messages = new CopyOnWriteArrayList<String>();
@@ -171,6 +189,7 @@ public class JdbcMessageStoreChannelOnePollerIntegrationTests {
 			}
 			return input;
 		}
+
 	}
 
 }

spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java

@@ -60,9 +60,7 @@ import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.simp.stomp.ReactorNettyTcpStompClient;
 import org.springframework.messaging.simp.stomp.StompCommand;
-import org.springframework.messaging.simp.stomp.StompReactorNettyCodec;
 import org.springframework.messaging.support.GenericMessage;
-import org.springframework.messaging.tcp.reactor.ReactorNettyTcpClient;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.util.SocketUtils;
 
@@ -97,12 +95,7 @@ public class StompServerIntegrationTests {
 		activeMQBroker.getSystemUsage().getTempUsage().setLimit(1024 * 1024 * 5);
 		activeMQBroker.start();
 
-
+		stompClient = new ReactorNettyTcpStompClient("127.0.0.1", port);
-// 	TODO until Reactor Bismuth-SR7	stompClient = new ReactorNettyTcpStompClient("127.0.0.1", port);
-		stompClient = new ReactorNettyTcpStompClient(
-				new ReactorNettyTcpClient<>(options -> options.disablePool().host("127.0.0.1").port(port),
-						new StompReactorNettyCodec()));
-
 		stompClient.setMessageConverter(new PassThruMessageConverter());
 		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
 		taskScheduler.afterPropertiesSet();