[SB3 Update] JDBC source and supplier updates

- Fix jdbc-source tests to use new Spring Integration Test Binder
- Fix jdbc-supplier test to check root cause

applications/source/file-source/README.adoc

@@ -25,35 +25,35 @@ Properties grouped by prefix:
 
 === file.consumer
 
-$$markers-json$$:: $$When 'fileMarkers == true', specify if they should be produced as FileSplitter.FileMarker objects or JSON.$$ *($$Boolean$$, default: `$$<none>$$`)*
+$$markers-json$$:: $$When 'fileMarkers == true', specify if they should be produced as FileSplitter.FileMarker objects or JSON.$$ *($$Boolean$$, default: `$$true$$`)*
 $$mode$$:: $$The FileReadingMode to use for file reading sources. Values are 'ref' - The File object, 'lines' - a message per line, or 'contents' - the contents as bytes.$$ *($$FileReadingMode$$, default: `$$<none>$$`, possible values: `ref`,`lines`,`contents`)*
 $$with-markers$$:: $$Set to true to emit start of file/end of file marker messages before/after the data. Only valid with FileReadingMode 'lines'.$$ *($$Boolean$$, default: `$$<none>$$`)*
 
 === file.supplier
 
-$$delay-when-empty$$:: $$Duration of delay when no new files are detected.$$ *($$Duration$$, default: `$$<none>$$`)*
+$$delay-when-empty$$:: $$Duration of delay when no new files are detected.$$ *($$Duration$$, default: `$$1s$$`)*
 $$directory$$:: $$The directory to poll for new files.$$ *($$File$$, default: `$$<none>$$`)*
 $$filename-pattern$$:: $$A simple ant pattern to match files.$$ *($$String$$, default: `$$<none>$$`)*
 $$filename-regex$$:: $$A regex pattern to match files.$$ *($$Pattern$$, default: `$$<none>$$`)*
-$$prevent-duplicates$$:: $$Set to true to include an AcceptOnceFileListFilter which prevents duplicates.$$ *($$Boolean$$, default: `$$<none>$$`)*
+$$prevent-duplicates$$:: $$Set to true to include an AcceptOnceFileListFilter which prevents duplicates.$$ *($$Boolean$$, default: `$$true$$`)*
 
 === metadata.store.dynamo-db
 
-$$create-delay$$:: $$Delay between create table retries.$$ *($$Integer$$, default: `$$<none>$$`)*
-$$create-retries$$:: $$Retry number for create table request.$$ *($$Integer$$, default: `$$<none>$$`)*
-$$read-capacity$$:: $$Read capacity on the table.$$ *($$Long$$, default: `$$<none>$$`)*
+$$create-delay$$:: $$Delay between create table retries.$$ *($$Integer$$, default: `$$1$$`)*
+$$create-retries$$:: $$Retry number for create table request.$$ *($$Integer$$, default: `$$25$$`)*
+$$read-capacity$$:: $$Read capacity on the table.$$ *($$Long$$, default: `$$1$$`)*
 $$table$$:: $$Table name for metadata.$$ *($$String$$, default: `$$<none>$$`)*
 $$time-to-live$$:: $$TTL for table entries.$$ *($$Integer$$, default: `$$<none>$$`)*
-$$write-capacity$$:: $$Write capacity on the table.$$ *($$Long$$, default: `$$<none>$$`)*
+$$write-capacity$$:: $$Write capacity on the table.$$ *($$Long$$, default: `$$1$$`)*
 
 === metadata.store.jdbc
 
-$$region$$:: $$Unique grouping identifier for messages persisted with this store.$$ *($$String$$, default: `$$<none>$$`)*
+$$region$$:: $$Unique grouping identifier for messages persisted with this store.$$ *($$String$$, default: `$$DEFAULT$$`)*
 $$table-prefix$$:: $$Prefix for the custom table name.$$ *($$String$$, default: `$$<none>$$`)*
 
 === metadata.store.mongo-db
 
-$$collection$$:: $$MongoDB collection name for metadata.$$ *($$String$$, default: `$$<none>$$`)*
+$$collection$$:: $$MongoDB collection name for metadata.$$ *($$String$$, default: `$$metadataStore$$`)*
 
 === metadata.store.redis
 
@@ -65,10 +65,10 @@ $$type$$:: $$Indicates the type of metadata store to configure (default is 'memo
 
 === metadata.store.zookeeper
 
-$$connect-string$$:: $$Zookeeper connect string in form HOST:PORT.$$ *($$String$$, default: `$$<none>$$`)*
-$$encoding$$:: $$Encoding to use when storing data in Zookeeper.$$ *($$Charset$$, default: `$$<none>$$`)*
-$$retry-interval$$:: $$Retry interval for Zookeeper operations in milliseconds.$$ *($$Integer$$, default: `$$<none>$$`)*
-$$root$$:: $$Root node - store entries are children of this node.$$ *($$String$$, default: `$$<none>$$`)*
+$$connect-string$$:: $$Zookeeper connect string in form HOST:PORT.$$ *($$String$$, default: `$$127.0.0.1:2181$$`)*
+$$encoding$$:: $$Encoding to use when storing data in Zookeeper.$$ *($$Charset$$, default: `$$UTF-8$$`)*
+$$retry-interval$$:: $$Retry interval for Zookeeper operations in milliseconds.$$ *($$Integer$$, default: `$$1000$$`)*
+$$root$$:: $$Root node - store entries are children of this node.$$ *($$String$$, default: `$$/SpringIntegration-MetadataStore$$`)*
 //end::configuration-properties[]
 
 //end::ref-doc[]

applications/source/ftp-source/README.adoc

@@ -73,7 +73,7 @@ Properties grouped by prefix:
 
 === file.consumer
 
-$$markers-json$$:: $$When 'fileMarkers == true', specify if they should be produced as FileSplitter.FileMarker objects or JSON.$$ *($$Boolean$$, default: `$$<none>$$`)*
+$$markers-json$$:: $$When 'fileMarkers == true', specify if they should be produced as FileSplitter.FileMarker objects or JSON.$$ *($$Boolean$$, default: `$$true$$`)*
 $$mode$$:: $$The FileReadingMode to use for file reading sources. Values are 'ref' - The File object, 'lines' - a message per line, or 'contents' - the contents as bytes.$$ *($$FileReadingMode$$, default: `$$<none>$$`, possible values: `ref`,`lines`,`contents`)*
 $$with-markers$$:: $$Set to true to emit start of file/end of file marker messages before/after the data. Only valid with FileReadingMode 'lines'.$$ *($$Boolean$$, default: `$$<none>$$`)*
 
@@ -81,41 +81,41 @@ $$with-markers$$:: $$Set to true to emit start of file/end of file marker messag
 
 $$cache-sessions$$:: $$Cache sessions.$$ *($$Boolean$$, default: `$$<none>$$`)*
 $$client-mode$$:: $$The client mode to use for the FTP session.$$ *($$ClientMode$$, default: `$$<none>$$`, possible values: `ACTIVE`,`PASSIVE`)*
-$$host$$:: $$The host name of the server.$$ *($$String$$, default: `$$<none>$$`)*
+$$host$$:: $$The host name of the server.$$ *($$String$$, default: `$$localhost$$`)*
 $$password$$:: $$The password to use to connect to the server.$$ *($$String$$, default: `$$<none>$$`)*
-$$port$$:: $$The port of the server.$$ *($$Integer$$, default: `$$<none>$$`)*
+$$port$$:: $$The port of the server.$$ *($$Integer$$, default: `$$21$$`)*
 $$username$$:: $$The username to use to connect to the server.$$ *($$String$$, default: `$$<none>$$`)*
 
 === ftp.supplier
 
-$$auto-create-local-dir$$:: $$Set to true to create the local directory if it does not exist.$$ *($$Boolean$$, default: `$$<none>$$`)*
-$$delay-when-empty$$:: $$Duration of delay when no new files are detected.$$ *($$Duration$$, default: `$$<none>$$`)*
-$$delete-remote-files$$:: $$Set to true to delete remote files after successful transfer.$$ *($$Boolean$$, default: `$$<none>$$`)*
+$$auto-create-local-dir$$:: $$Set to true to create the local directory if it does not exist.$$ *($$Boolean$$, default: `$$true$$`)*
+$$delay-when-empty$$:: $$Duration of delay when no new files are detected.$$ *($$Duration$$, default: `$$1s$$`)*
+$$delete-remote-files$$:: $$Set to true to delete remote files after successful transfer.$$ *($$Boolean$$, default: `$$false$$`)*
 $$filename-pattern$$:: $$A filter pattern to match the names of files to transfer.$$ *($$String$$, default: `$$<none>$$`)*
 $$filename-regex$$:: $$A filter regex pattern to match the names of files to transfer.$$ *($$Pattern$$, default: `$$<none>$$`)*
 $$local-dir$$:: $$The local directory to use for file transfers.$$ *($$File$$, default: `$$<none>$$`)*
-$$preserve-timestamp$$:: $$Set to true to preserve the original timestamp.$$ *($$Boolean$$, default: `$$<none>$$`)*
-$$remote-dir$$:: $$The remote FTP directory.$$ *($$String$$, default: `$$<none>$$`)*
-$$remote-file-separator$$:: $$The remote file separator.$$ *($$String$$, default: `$$<none>$$`)*
-$$tmp-file-suffix$$:: $$The suffix to use while the transfer is in progress.$$ *($$String$$, default: `$$<none>$$`)*
+$$preserve-timestamp$$:: $$Set to true to preserve the original timestamp.$$ *($$Boolean$$, default: `$$true$$`)*
+$$remote-dir$$:: $$The remote FTP directory.$$ *($$String$$, default: `$$/$$`)*
+$$remote-file-separator$$:: $$The remote file separator.$$ *($$String$$, default: `$$/$$`)*
+$$tmp-file-suffix$$:: $$The suffix to use while the transfer is in progress.$$ *($$String$$, default: `$$.tmp$$`)*
 
 === metadata.store.dynamo-db
 
-$$create-delay$$:: $$Delay between create table retries.$$ *($$Integer$$, default: `$$<none>$$`)*
-$$create-retries$$:: $$Retry number for create table request.$$ *($$Integer$$, default: `$$<none>$$`)*
-$$read-capacity$$:: $$Read capacity on the table.$$ *($$Long$$, default: `$$<none>$$`)*
+$$create-delay$$:: $$Delay between create table retries.$$ *($$Integer$$, default: `$$1$$`)*
+$$create-retries$$:: $$Retry number for create table request.$$ *($$Integer$$, default: `$$25$$`)*
+$$read-capacity$$:: $$Read capacity on the table.$$ *($$Long$$, default: `$$1$$`)*
 $$table$$:: $$Table name for metadata.$$ *($$String$$, default: `$$<none>$$`)*
 $$time-to-live$$:: $$TTL for table entries.$$ *($$Integer$$, default: `$$<none>$$`)*
-$$write-capacity$$:: $$Write capacity on the table.$$ *($$Long$$, default: `$$<none>$$`)*
+$$write-capacity$$:: $$Write capacity on the table.$$ *($$Long$$, default: `$$1$$`)*
 
 === metadata.store.jdbc
 
-$$region$$:: $$Unique grouping identifier for messages persisted with this store.$$ *($$String$$, default: `$$<none>$$`)*
+$$region$$:: $$Unique grouping identifier for messages persisted with this store.$$ *($$String$$, default: `$$DEFAULT$$`)*
 $$table-prefix$$:: $$Prefix for the custom table name.$$ *($$String$$, default: `$$<none>$$`)*
 
 === metadata.store.mongo-db
 
-$$collection$$:: $$MongoDB collection name for metadata.$$ *($$String$$, default: `$$<none>$$`)*
+$$collection$$:: $$MongoDB collection name for metadata.$$ *($$String$$, default: `$$metadataStore$$`)*
 
 === metadata.store.redis
 
@@ -127,10 +127,10 @@ $$type$$:: $$Indicates the type of metadata store to configure (default is 'memo
 
 === metadata.store.zookeeper
 
-$$connect-string$$:: $$Zookeeper connect string in form HOST:PORT.$$ *($$String$$, default: `$$<none>$$`)*
-$$encoding$$:: $$Encoding to use when storing data in Zookeeper.$$ *($$Charset$$, default: `$$<none>$$`)*
-$$retry-interval$$:: $$Retry interval for Zookeeper operations in milliseconds.$$ *($$Integer$$, default: `$$<none>$$`)*
-$$root$$:: $$Root node - store entries are children of this node.$$ *($$String$$, default: `$$<none>$$`)*
+$$connect-string$$:: $$Zookeeper connect string in form HOST:PORT.$$ *($$String$$, default: `$$127.0.0.1:2181$$`)*
+$$encoding$$:: $$Encoding to use when storing data in Zookeeper.$$ *($$Charset$$, default: `$$UTF-8$$`)*
+$$retry-interval$$:: $$Retry interval for Zookeeper operations in milliseconds.$$ *($$Integer$$, default: `$$1000$$`)*
+$$root$$:: $$Root node - store entries are children of this node.$$ *($$String$$, default: `$$/SpringIntegration-MetadataStore$$`)*
 //end::configuration-properties[]
 
 == Examples

applications/source/jdbc-source/pom.xml

@@ -19,11 +19,13 @@
             <groupId>org.springframework.cloud.fn</groupId>
             <artifactId>jdbc-supplier</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-stream-test-support</artifactId>
-            <scope>test</scope>
-        </dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream</artifactId>
+			<type>test-jar</type>
+			<scope>test</scope>
+			<classifier>test-binder</classifier>
+		</dependency>
         <dependency>
             <groupId>com.h2database</groupId>
             <artifactId>h2</artifactId>

applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/DefaultBehaviorTests.java

@@ -17,7 +17,6 @@
 package org.springframework.cloud.stream.app.source.jdbc;
 
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.Test;
 
@@ -26,31 +25,25 @@ import org.springframework.test.context.TestPropertySource;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+/**
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
 @TestPropertySource(properties = "jdbc.supplier.query=select id, name from test order by id")
 class DefaultBehaviorTests extends JdbcSourceIntegrationTests {
 
 	@Test
-	void testExtraction() throws Exception {
-		Message<?> received = messageCollector.forChannel(output).poll(10, TimeUnit.SECONDS);
-		assertThat(received).isNotNull();
-		assertThat(received.getPayload().getClass()).isEqualTo(String.class);
-
-		Map<?, ?> payload = this.objectMapper.readValue((String) received.getPayload(), Map.class);
-
+	void testExtraction() {
+		Message<?> received = receiveMessage(10_000);
+		Map<?, ?> payload = extractPayload(received, Map.class);
 		assertThat(payload.get("ID")).isEqualTo(1);
-		received = messageCollector.forChannel(output).poll(10, TimeUnit.SECONDS);
-		assertThat(received).isNotNull();
-		assertThat(received.getPayload().getClass()).isEqualTo(String.class);
-
-		payload = this.objectMapper.readValue((String) received.getPayload(), Map.class);
 
+		received = receiveMessage(10_000);
+		payload = extractPayload(received, Map.class);
 		assertThat(payload.get("ID")).isEqualTo(2);
-		received = messageCollector.forChannel(output).poll(10, TimeUnit.SECONDS);
-		assertThat(received).isNotNull();
-		assertThat(received.getPayload().getClass()).isEqualTo(String.class);
-
-		payload = this.objectMapper.readValue((String) received.getPayload(), Map.class);
 
+		received = receiveMessage(10_000);
+		payload = extractPayload(received, Map.class);
 		assertThat(payload.get("ID")).isEqualTo(3);
 	}
 }

applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/JdbcSourceIntegrationTests.java

@@ -16,43 +16,74 @@
 
 package org.springframework.cloud.stream.app.source.jdbc;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.cloud.fn.supplier.jdbc.JdbcSupplierConfiguration;
-import org.springframework.cloud.stream.test.binder.MessageCollector;
+import org.springframework.cloud.stream.binder.test.OutputDestination;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
 import org.springframework.context.annotation.Import;
 import org.springframework.jdbc.core.JdbcOperations;
-import org.springframework.messaging.MessageChannel;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
 import org.springframework.test.annotation.DirtiesContext;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 /**
  * @author Soby Chacko
  * @author Artem Bilan
+ * @author Chris Bono
  */
 @SpringBootTest(properties = "spring.cloud.function.definition=jdbcSupplier")
 @DirtiesContext
 public class JdbcSourceIntegrationTests {
 
 	@Autowired
-	protected ObjectMapper objectMapper;
-
-	@Qualifier("jdbcSupplier-out-0")
-	@Autowired
-	protected MessageChannel output;
+	private ObjectMapper objectMapper;
 
 	@Autowired
 	protected JdbcOperations jdbcOperations;
 
 	@Autowired
-	protected MessageCollector messageCollector;
+	private OutputDestination outputDestination;
 
-	@SpringBootApplication
-	@Import(JdbcSupplierConfiguration.class)
-	public static class JdbcSourceTestApplication {
+	@Nullable
+	protected Message<?> receiveMessageMaybeNull(long timeoutMillis) {
+		return this.outputDestination.receive(timeoutMillis, "jdbcSupplier-out-0");
 	}
 
+	protected Message<?> receiveMessage(long timeoutMillis) {
+		Message<?> received = this.outputDestination.receive(timeoutMillis, "jdbcSupplier-out-0");
+		assertThat(received).isNotNull();
+		assertThat(received.getPayload().getClass()).isEqualTo(byte[].class);
+		return received;
+	}
+
+	protected <T> T extractPayload(Message<?> message, Class<T> type) {
+		try {
+			return this.objectMapper.readValue(new String((byte[]) message.getPayload()), type);
+		}
+		catch (JsonProcessingException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	protected <T> T extractPayload(Message<?> message, JavaType type) {
+		try {
+			return this.objectMapper.readValue(new String((byte[]) message.getPayload()), type);
+		}
+		catch (JsonProcessingException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	@SpringBootApplication
+	@Import({ JdbcSupplierConfiguration.class, TestChannelBinderConfiguration.class })
+	public static class JdbcSourceTestApplication {
+	}
 }

applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/Select2PerPollNoSplitWithUpdateTests.java

@@ -18,7 +18,6 @@ package org.springframework.cloud.stream.app.source.jdbc;
 
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
 
 import com.fasterxml.jackson.databind.type.CollectionLikeType;
 import com.fasterxml.jackson.databind.type.TypeFactory;
@@ -32,6 +31,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 /**
  * @author Soby Chacko
  * @author Artem Bilan
+ * @author Chris Bono
  */
 @TestPropertySource(properties = {
 		"jdbc.supplier.query=select id, name, tag from test where tag is NULL order by id",
@@ -41,22 +41,18 @@ import static org.assertj.core.api.Assertions.assertThat;
 public class Select2PerPollNoSplitWithUpdateTests extends JdbcSourceIntegrationTests {
 
 	@Test
-	public void testExtraction() throws Exception {
-		Message<?> received = this.messageCollector.forChannel(this.output).poll(10, TimeUnit.SECONDS);
-		assertThat(received).isNotNull();
-		assertThat(received.getPayload().getClass()).isEqualTo(String.class);
-
+	public void testExtraction() {
 		CollectionLikeType valueType = TypeFactory.defaultInstance()
 				.constructCollectionLikeType(List.class, Map.class);
 
-		List<Map<?, ?>> payload = this.objectMapper.readValue((String) received.getPayload(), valueType);
-
+		Message<?> received = receiveMessage(10_000);
+		List<Map<?, ?>> payload = extractPayload(received, valueType);
 		assertThat(payload.size()).isEqualTo(2);
 		assertThat(payload.get(0).get("ID")).isEqualTo(1);
 		assertThat(payload.get(1).get("ID")).isEqualTo(2);
-		received = this.messageCollector.forChannel(this.output).poll(10, TimeUnit.SECONDS);
-		assertThat(received).isNotNull();
-		payload = this.objectMapper.readValue((String) received.getPayload(), valueType);
+
+		received = receiveMessage(10_000);
+		payload = extractPayload(received, valueType);
 		assertThat(payload.size()).isEqualTo(1);
 		assertThat(payload.get(0).get("ID")).isEqualTo(3);
 	}

applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/SelectAllNoSplitTests.java

@@ -18,7 +18,6 @@ package org.springframework.cloud.stream.app.source.jdbc;
 
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
 
 import com.fasterxml.jackson.databind.type.CollectionLikeType;
 import com.fasterxml.jackson.databind.type.TypeFactory;
@@ -32,6 +31,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 /**
  * @author Soby Chacko
  * @author Artem Bilan
+ * @author Chris Bono
  */
 @TestPropertySource(properties = {
 		"jdbc.supplier.query=select id, name, tag from test where tag is NULL order by id",
@@ -40,16 +40,11 @@ import static org.assertj.core.api.Assertions.assertThat;
 public class SelectAllNoSplitTests extends JdbcSourceIntegrationTests {
 
 	@Test
-	public void testExtraction() throws Exception {
-		Message<?> received = messageCollector.forChannel(output).poll(10, TimeUnit.SECONDS);
-		assertThat(received).isNotNull();
-		assertThat(received.getPayload().getClass()).isEqualTo(String.class);
-
+	public void testExtraction() {
 		CollectionLikeType valueType = TypeFactory.defaultInstance()
 				.constructCollectionLikeType(List.class, Map.class);
-
-		List<Map<?, ?>> payload = this.objectMapper.readValue((String) received.getPayload(), valueType);
-
+		Message<?> received = receiveMessage(10_000);
+		List<Map<?, ?>> payload = extractPayload(received, valueType);
 		assertThat(payload.size()).isEqualTo(3);
 		assertThat(payload.get(0).get("ID")).isEqualTo(1);
 		assertThat(payload.get(2).get("NAME")).isEqualTo("John");

applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/SelectAllWithDelayTests.java

@@ -17,7 +17,6 @@
 package org.springframework.cloud.stream.app.source.jdbc;
 
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.Test;
 
@@ -26,32 +25,29 @@ import org.springframework.test.context.TestPropertySource;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+/**
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
 @TestPropertySource(properties = {"jdbc.supplier.query=select id, name from test order by id", "spring.cloud.stream.poller.fixedDelay=60000"})
 public class SelectAllWithDelayTests extends JdbcSourceIntegrationTests {
 
 	@Test
-	public void testExtraction() throws Exception {
-		Message<?> received = messageCollector.forChannel(output).poll(10, TimeUnit.SECONDS);
-		assertThat(received).isNotNull();
-		assertThat(received.getPayload().getClass()).isEqualTo(String.class);
-
-		Map<?, ?> payload = this.objectMapper.readValue((String) received.getPayload(), Map.class);
-
+	public void testExtraction() {
+		Message<?> received = receiveMessage(10_000);
+		Map<?, ?> payload = extractPayload(received, Map.class);
 		assertThat(payload.get("ID")).isEqualTo(1);
-		received = messageCollector.forChannel(output).poll(10, TimeUnit.SECONDS);
-		assertThat(received).isNotNull();
-		assertThat(received.getPayload().getClass()).isEqualTo(String.class);
 
-		payload = this.objectMapper.readValue((String) received.getPayload(), Map.class);
+		received = receiveMessage(10_000);
+		payload = extractPayload(received, Map.class);
 		assertThat(payload.get("ID")).isEqualTo(2);
-		received = messageCollector.forChannel(output).poll(10, TimeUnit.SECONDS);
-		assertThat(received).isNotNull();
-		assertThat(received.getPayload().getClass()).isEqualTo(String.class);
 
-		payload = this.objectMapper.readValue((String) received.getPayload(), Map.class);
+		received = receiveMessage(10_000);
+		payload = extractPayload(received, Map.class);
 		assertThat(payload.get("ID")).isEqualTo(3);
+
 		// should not wrap around to the beginning since delay is 60
-		received = messageCollector.forChannel(output).poll(1, TimeUnit.SECONDS);
+		received = receiveMessageMaybeNull(10_000);
 		assertThat(received).isNull();
 	}
 }

applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/SelectAllWithMinDelayTests.java

@@ -17,7 +17,6 @@
 package org.springframework.cloud.stream.app.source.jdbc;
 
 import java.util.Map;
-import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.Test;
 
@@ -26,38 +25,33 @@ import org.springframework.test.context.TestPropertySource;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-@TestPropertySource(properties = {"jdbc.supplier.query=select id, name from test order by id", "spring.cloud.stream.poller.fixedDelay=1"})
+/**
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+@TestPropertySource(properties = {
+		"jdbc.supplier.query=select id, name from test order by id",
+		"spring.cloud.stream.poller.fixedDelay=1"
+})
 public class SelectAllWithMinDelayTests extends JdbcSourceIntegrationTests {
 
 	@Test
-	public void testExtraction() throws Exception {
-		Message<?> received = messageCollector.forChannel(output).poll(10, TimeUnit.SECONDS);
-		assertThat(received).isNotNull();
-		assertThat(received.getPayload().getClass()).isEqualTo(String.class);
-
-		Map<?, ?> payload = this.objectMapper.readValue((String) received.getPayload(), Map.class);
-
+	public void testExtraction() {
+		Message<?> received = receiveMessage(10_000);
+		Map<?, ?> payload = extractPayload(received, Map.class);
 		assertThat(payload.get("ID")).isEqualTo(1);
-		received = messageCollector.forChannel(output).poll(10, TimeUnit.SECONDS);
-		assertThat(received).isNotNull();
-		assertThat(received.getPayload().getClass()).isEqualTo(String.class);
-		payload = this.objectMapper.readValue((String) received.getPayload(), Map.class);
 
+		received = receiveMessage(10_000);
+		payload = extractPayload(received, Map.class);
 		assertThat(payload.get("ID")).isEqualTo(2);
-		received = messageCollector.forChannel(output).poll(10, TimeUnit.SECONDS);
-		assertThat(received).isNotNull();
-		assertThat(received.getPayload().getClass()).isEqualTo(String.class);
-
-		payload = this.objectMapper.readValue((String) received.getPayload(), Map.class);
 
+		received = receiveMessage(10_000);
+		payload = extractPayload(received, Map.class);
 		assertThat(payload.get("ID")).isEqualTo(3);
+
 		// should wrap around to the beginning
-		received = messageCollector.forChannel(output).poll(2, TimeUnit.SECONDS);
-		assertThat(received).isNotNull();
-		assertThat(received.getPayload().getClass()).isEqualTo(String.class);
-
-		payload = this.objectMapper.readValue((String) received.getPayload(), Map.class);
-
+		received = receiveMessage(10_000);
+		payload = extractPayload(received, Map.class);
 		assertThat(payload.get("ID")).isEqualTo(1);
 	}
 

functions/supplier/jdbc-supplier/src/test/java/org/springframework/cloud/fn/supplier/jdbc/DefaultJdbcSupplierTests.java

@@ -91,6 +91,7 @@ public class DefaultJdbcSupplierTests {
 	void verifyNoIntMessageGroupTable() {
 		assertThatExceptionOfType(BadSqlGrammarException.class)
 				.isThrownBy(() -> this.jdbcTemplate.queryForList("SELECT * FROM INT_MESSAGE_GROUP"))
+				.havingCause()
 				.withMessageContaining("Table \"INT_MESSAGE_GROUP\" not found;");
 	}