From c0b48ebb0cc68fd88db2bf7176bf0fb4274f4cc7 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Sun, 16 Oct 2022 23:23:01 -0500 Subject: [PATCH] [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 | 26 ++++----- applications/source/ftp-source/README.adoc | 40 +++++++------- applications/source/jdbc-source/pom.xml | 12 ++-- .../app/source/jdbc/DefaultBehaviorTests.java | 29 ++++------ .../jdbc/JdbcSourceIntegrationTests.java | 55 +++++++++++++++---- .../Select2PerPollNoSplitWithUpdateTests.java | 18 +++--- .../source/jdbc/SelectAllNoSplitTests.java | 13 ++--- .../source/jdbc/SelectAllWithDelayTests.java | 30 +++++----- .../jdbc/SelectAllWithMinDelayTests.java | 42 ++++++-------- .../jdbc/DefaultJdbcSupplierTests.java | 1 + 10 files changed, 137 insertions(+), 129 deletions(-) diff --git a/applications/source/file-source/README.adoc b/applications/source/file-source/README.adoc index 2a7d196b..d4966c19 100644 --- a/applications/source/file-source/README.adoc +++ b/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: `$$$$`)* +$$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: `$$$$`, 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: `$$$$`)* === file.supplier -$$delay-when-empty$$:: $$Duration of delay when no new files are detected.$$ *($$Duration$$, default: `$$$$`)* +$$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: `$$$$`)* $$filename-pattern$$:: $$A simple ant pattern to match files.$$ *($$String$$, default: `$$$$`)* $$filename-regex$$:: $$A regex pattern to match files.$$ *($$Pattern$$, default: `$$$$`)* -$$prevent-duplicates$$:: $$Set to true to include an AcceptOnceFileListFilter which prevents duplicates.$$ *($$Boolean$$, default: `$$$$`)* +$$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: `$$$$`)* -$$create-retries$$:: $$Retry number for create table request.$$ *($$Integer$$, default: `$$$$`)* -$$read-capacity$$:: $$Read capacity on the table.$$ *($$Long$$, default: `$$$$`)* +$$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: `$$$$`)* $$time-to-live$$:: $$TTL for table entries.$$ *($$Integer$$, default: `$$$$`)* -$$write-capacity$$:: $$Write capacity on the table.$$ *($$Long$$, default: `$$$$`)* +$$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: `$$$$`)* +$$region$$:: $$Unique grouping identifier for messages persisted with this store.$$ *($$String$$, default: `$$DEFAULT$$`)* $$table-prefix$$:: $$Prefix for the custom table name.$$ *($$String$$, default: `$$$$`)* === metadata.store.mongo-db -$$collection$$:: $$MongoDB collection name for metadata.$$ *($$String$$, default: `$$$$`)* +$$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: `$$$$`)* -$$encoding$$:: $$Encoding to use when storing data in Zookeeper.$$ *($$Charset$$, default: `$$$$`)* -$$retry-interval$$:: $$Retry interval for Zookeeper operations in milliseconds.$$ *($$Integer$$, default: `$$$$`)* -$$root$$:: $$Root node - store entries are children of this node.$$ *($$String$$, default: `$$$$`)* +$$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[] diff --git a/applications/source/ftp-source/README.adoc b/applications/source/ftp-source/README.adoc index 305d16fb..702e3309 100644 --- a/applications/source/ftp-source/README.adoc +++ b/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: `$$$$`)* +$$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: `$$$$`, 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: `$$$$`)* @@ -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: `$$$$`)* $$client-mode$$:: $$The client mode to use for the FTP session.$$ *($$ClientMode$$, default: `$$$$`, possible values: `ACTIVE`,`PASSIVE`)* -$$host$$:: $$The host name of the server.$$ *($$String$$, default: `$$$$`)* +$$host$$:: $$The host name of the server.$$ *($$String$$, default: `$$localhost$$`)* $$password$$:: $$The password to use to connect to the server.$$ *($$String$$, default: `$$$$`)* -$$port$$:: $$The port of the server.$$ *($$Integer$$, default: `$$$$`)* +$$port$$:: $$The port of the server.$$ *($$Integer$$, default: `$$21$$`)* $$username$$:: $$The username to use to connect to the server.$$ *($$String$$, default: `$$$$`)* === ftp.supplier -$$auto-create-local-dir$$:: $$Set to true to create the local directory if it does not exist.$$ *($$Boolean$$, default: `$$$$`)* -$$delay-when-empty$$:: $$Duration of delay when no new files are detected.$$ *($$Duration$$, default: `$$$$`)* -$$delete-remote-files$$:: $$Set to true to delete remote files after successful transfer.$$ *($$Boolean$$, default: `$$$$`)* +$$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: `$$$$`)* $$filename-regex$$:: $$A filter regex pattern to match the names of files to transfer.$$ *($$Pattern$$, default: `$$$$`)* $$local-dir$$:: $$The local directory to use for file transfers.$$ *($$File$$, default: `$$$$`)* -$$preserve-timestamp$$:: $$Set to true to preserve the original timestamp.$$ *($$Boolean$$, default: `$$$$`)* -$$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: `$$$$`)* +$$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: `$$$$`)* -$$create-retries$$:: $$Retry number for create table request.$$ *($$Integer$$, default: `$$$$`)* -$$read-capacity$$:: $$Read capacity on the table.$$ *($$Long$$, default: `$$$$`)* +$$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: `$$$$`)* $$time-to-live$$:: $$TTL for table entries.$$ *($$Integer$$, default: `$$$$`)* -$$write-capacity$$:: $$Write capacity on the table.$$ *($$Long$$, default: `$$$$`)* +$$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: `$$$$`)* +$$region$$:: $$Unique grouping identifier for messages persisted with this store.$$ *($$String$$, default: `$$DEFAULT$$`)* $$table-prefix$$:: $$Prefix for the custom table name.$$ *($$String$$, default: `$$$$`)* === metadata.store.mongo-db -$$collection$$:: $$MongoDB collection name for metadata.$$ *($$String$$, default: `$$$$`)* +$$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: `$$$$`)* -$$encoding$$:: $$Encoding to use when storing data in Zookeeper.$$ *($$Charset$$, default: `$$$$`)* -$$retry-interval$$:: $$Retry interval for Zookeeper operations in milliseconds.$$ *($$Integer$$, default: `$$$$`)* -$$root$$:: $$Root node - store entries are children of this node.$$ *($$String$$, default: `$$$$`)* +$$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 diff --git a/applications/source/jdbc-source/pom.xml b/applications/source/jdbc-source/pom.xml index fc53cdea..ba005e5f 100644 --- a/applications/source/jdbc-source/pom.xml +++ b/applications/source/jdbc-source/pom.xml @@ -19,11 +19,13 @@ org.springframework.cloud.fn jdbc-supplier - - org.springframework.cloud - spring-cloud-stream-test-support - test - + + org.springframework.cloud + spring-cloud-stream + test-jar + test + test-binder + com.h2database h2 diff --git a/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/DefaultBehaviorTests.java b/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/DefaultBehaviorTests.java index b5f9cd79..d53768c4 100644 --- a/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/DefaultBehaviorTests.java +++ b/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); } } diff --git a/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/JdbcSourceIntegrationTests.java b/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/JdbcSourceIntegrationTests.java index 4160d269..dca81b83 100644 --- a/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/JdbcSourceIntegrationTests.java +++ b/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 extractPayload(Message message, Class type) { + try { + return this.objectMapper.readValue(new String((byte[]) message.getPayload()), type); + } + catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + } + + protected 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 { + } } diff --git a/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/Select2PerPollNoSplitWithUpdateTests.java b/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/Select2PerPollNoSplitWithUpdateTests.java index 7f40146c..86112467 100644 --- a/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/Select2PerPollNoSplitWithUpdateTests.java +++ b/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> payload = this.objectMapper.readValue((String) received.getPayload(), valueType); - + Message received = receiveMessage(10_000); + List> 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); } diff --git a/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/SelectAllNoSplitTests.java b/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/SelectAllNoSplitTests.java index fb9d3838..3f09a7d0 100644 --- a/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/SelectAllNoSplitTests.java +++ b/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> payload = this.objectMapper.readValue((String) received.getPayload(), valueType); - + Message received = receiveMessage(10_000); + List> 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"); diff --git a/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/SelectAllWithDelayTests.java b/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/SelectAllWithDelayTests.java index b89dd4b3..5707afec 100644 --- a/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/SelectAllWithDelayTests.java +++ b/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(); } } diff --git a/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/SelectAllWithMinDelayTests.java b/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/SelectAllWithMinDelayTests.java index 099be995..db0a50aa 100644 --- a/applications/source/jdbc-source/src/test/java/org/springframework/cloud/stream/app/source/jdbc/SelectAllWithMinDelayTests.java +++ b/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); } diff --git a/functions/supplier/jdbc-supplier/src/test/java/org/springframework/cloud/fn/supplier/jdbc/DefaultJdbcSupplierTests.java b/functions/supplier/jdbc-supplier/src/test/java/org/springframework/cloud/fn/supplier/jdbc/DefaultJdbcSupplierTests.java index 5a4684a4..44073f55 100644 --- a/functions/supplier/jdbc-supplier/src/test/java/org/springframework/cloud/fn/supplier/jdbc/DefaultJdbcSupplierTests.java +++ b/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;"); }