Fix JdbcMessageStoreChannelTests race condition spring-projects#2

The message in the `QueueChannel` appears for consuming a bit earlier
than TX is committed

* Introduce `afterCommitLatch` into the test verify the state
when TX is really committed and data is removed from DB

Author: artembilan

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

@@ -22,22 +22,26 @@
 		<int:queue ref="queue"/>
 	</int:channel>
 
-	<int:channel id="output">
-		<int:queue/>
-	</int:channel>
-
 	<bean id="queue" class="org.springframework.integration.store.MessageGroupQueue">
 		<constructor-arg ref="messageStore" />
 		<constructor-arg value="JdbcMessageStoreChannelTests" />
 	</bean>
 
-	<int:service-activator id="service-activator" input-channel="input" output-channel="output">
+	<int:service-activator id="service-activator" input-channel="input" output-channel="nullChannel">
 		<beans:bean class="org.springframework.integration.jdbc.store.JdbcMessageStoreChannelTests$Service" />
 		<int:poller fixed-rate="2000">
-			<int:transactional />
+			<int:transactional synchronization-factory="transactionSynchronizationFactory"/>
 		</int:poller>
 	</int:service-activator>
 
+	<int:transaction-synchronization-factory id="transactionSynchronizationFactory">
+		<int:after-commit expression="@afterCommitLatch.countDown()"/>
+	</int:transaction-synchronization-factory>
+
+	<bean id="afterCommitLatch" class="java.util.concurrent.CountDownLatch">
+		<constructor-arg value="1"/>
+	</bean>
+
 	<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
 		<property name="dataSource" ref="dataSource" />
 	</bean>

spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/JdbcMessageStoreChannelTests.java

@@ -18,6 +18,7 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 import java.util.List;
@@ -61,7 +62,7 @@ public class JdbcMessageStoreChannelTests {
 	private MessageChannel input;
 
 	@Autowired
-	private PollableChannel output;
+	private CountDownLatch afterCommitLatch;
 
 	@Autowired
 	private JdbcMessageStore messageStore;
@@ -70,6 +71,8 @@ public class JdbcMessageStoreChannelTests {
 	@Qualifier("service-activator")
 	private AbstractEndpoint serviceActivator;
 
+
+
 	@Before
 	public void init() {
 		Service.reset(1);
@@ -89,9 +92,9 @@ public void clear() {
 	}
 
 	@Test
-	public void testSendAndActivate() {
+	public void testSendAndActivate() throws InterruptedException {
 		this.input.send(new GenericMessage<>("foo"));
-		assertNotNull(output.receive(10000));
+		assertTrue(this.afterCommitLatch.await(10, TimeUnit.SECONDS));
 		assertEquals(1, Service.messages.size());
 		assertEquals(0, messageStore.getMessageGroup("JdbcMessageStoreChannelTests").size());
 	}