From a7a37950e94bc79ff0c0c71afa44730529f5fbd9 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 18 Sep 2020 19:59:54 -0400 Subject: [PATCH] Migrating pgcopy-sink * pgcopy-sink is migrated from Spring Cloud Stream App Starters repository. * This is not converted as a function, rather preserved as a sink app. * Old location: https://github.com/spring-cloud-stream-app-starters/jdbc/tree/master/spring-cloud-starter-stream-sink-pgcopy --- applications/sink/pgcopy-sink/README.adoc | 58 +++ applications/sink/pgcopy-sink/pom.xml | 139 ++++++++ .../DefaultInitializationScriptResource.java | 59 ++++ .../pgcopy/sink/PgcopySinkConfiguration.java | 332 ++++++++++++++++++ .../app/pgcopy/sink/PgcopySinkProperties.java | 204 +++++++++++ ...onfiguration-metadata-whitelist.properties | 6 + ...onfiguration-metadata-whitelist.properties | 6 + .../main/resources/META-INF/spring.provides | 1 + .../PgcopyBadErrorTableIntegrationTests.java | 170 +++++++++ .../PgcopyErrorTableIntegrationTests.java | 80 +++++ .../sink/PgcopySinkIntegrationTests.java | 197 +++++++++++ .../sink/PgcopySinkPropertiesTests.java | 153 ++++++++ .../app/pgcopy/test/PostgresTestSupport.java | 66 ++++ .../src/test/resources/application.properties | 4 + .../src/test/resources/error-table-ddl.sql | 2 + applications/sink/pom.xml | 1 + 16 files changed, 1478 insertions(+) create mode 100644 applications/sink/pgcopy-sink/README.adoc create mode 100644 applications/sink/pgcopy-sink/pom.xml create mode 100644 applications/sink/pgcopy-sink/src/main/java/org/springframework/cloud/stream/app/pgcopy/sink/DefaultInitializationScriptResource.java create mode 100644 applications/sink/pgcopy-sink/src/main/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkConfiguration.java create mode 100644 applications/sink/pgcopy-sink/src/main/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkProperties.java create mode 100644 applications/sink/pgcopy-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties create mode 100644 applications/sink/pgcopy-sink/src/main/resources/META-INF/spring-configuration-metadata-whitelist.properties create mode 100644 applications/sink/pgcopy-sink/src/main/resources/META-INF/spring.provides create mode 100644 applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopyBadErrorTableIntegrationTests.java create mode 100644 applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopyErrorTableIntegrationTests.java create mode 100644 applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkIntegrationTests.java create mode 100644 applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkPropertiesTests.java create mode 100644 applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/test/PostgresTestSupport.java create mode 100644 applications/sink/pgcopy-sink/src/test/resources/application.properties create mode 100644 applications/sink/pgcopy-sink/src/test/resources/error-table-ddl.sql diff --git a/applications/sink/pgcopy-sink/README.adoc b/applications/sink/pgcopy-sink/README.adoc new file mode 100644 index 00000000..849c3fb1 --- /dev/null +++ b/applications/sink/pgcopy-sink/README.adoc @@ -0,0 +1,58 @@ +//tag::ref-doc[] += Pgcopy Sink + +A module that writes its incoming payload to an RDBMS using the PostgreSQL COPY command. + +== Input + +=== Headers + +=== Payload + +* Any + +Column expression will be evaluated against the message and the expression will usually be compatible with only one type (such as a Map or bean etc.) + +== Output + +N/A + +== Options + +The **$$jdbc$$** $$sink$$ has the following options: + +//tag::configuration-properties[] +$$spring.datasource.driver-class-name$$:: $$Fully qualified name of the JDBC driver. Auto-detected based on the URL by default.$$ *($$String$$, default: `$$$$`)* +$$spring.datasource.password$$:: $$Login password of the database.$$ *($$String$$, default: `$$$$`)* +$$spring.datasource.url$$:: $$JDBC URL of the database.$$ *($$String$$, default: `$$$$`)* +$$spring.datasource.username$$:: $$Login username of the database.$$ *($$String$$, default: `$$$$`)* +//end::configuration-properties[] + +NOTE: The module also uses Spring Boot's https://docs.spring.io/spring-boot/docs/current/reference/html/boot-features-sql.html#boot-features-configure-datasource[DataSource support] for configuring the database connection, so properties like `spring.datasource.url` _etc._ apply. + +== Build + +``` +$ ./mvnw clean install -PgenerateApps +$ cd apps +``` +You can find the corresponding binder based projects here. +You can then cd into one one of the folders and build it: +``` +$ ./mvnw clean package +``` + +For integration tests to run, start a PostgreSQL database on localhost: + +``` + docker run -e POSTGRES_PASSWORD=spring -e POSTGRES_DB=test -p 5432:5432 -d postgres:latest +``` + +== Examples + +``` +java -jar pgcopy-sink.jar --tableName=names --columns=name --spring.datasource.driver-class-name=org.mariadb.jdbc.Driver \ +--spring.datasource.url='jdbc:mysql://localhost:3306/test +``` + +//end::ref-doc[] diff --git a/applications/sink/pgcopy-sink/pom.xml b/applications/sink/pgcopy-sink/pom.xml new file mode 100644 index 00000000..f136fff4 --- /dev/null +++ b/applications/sink/pgcopy-sink/pom.xml @@ -0,0 +1,139 @@ + + + 4.0.0 + pgcopy-sink + 3.0.0-SNAPSHOT + pgcopy-sink + pgcopy sink apps + jar + + + org.springframework.cloud.stream.app + stream-applications-core + 3.0.0-SNAPSHOT + + + + + + org.springframework.boot + spring-boot-starter-jdbc + + + org.springframework.integration + spring-integration-jdbc + + + org.postgresql + postgresql + compile + + + org.springframework.cloud + spring-cloud-stream-test-support-internal + ${spring-cloud-stream.version} + test + + + + + + + org.springframework.cloud + spring-cloud-dataflow-apps-docs-plugin + + + org.springframework.cloud + spring-cloud-dataflow-apps-generator-plugin + + + rabbit + sink + ${project.version} + org.springframework.cloud.fn.consumer.rabbit.RabbitConsumerConfiguration.class + + + + + org.springframework.cloud.stream.app + pgcopy-sink + ${project.version} + + + + + + + + + + + + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + + false + + spring-releases + Spring Releases + https://repo.spring.io/release + + + + false + + spring-libs-release + Spring Libs Release + https://repo.spring.io/libs-release + + + + false + + spring-milestone-release + Spring Milestone Release + https://repo.spring.io/libs-milestone + + + + + spring-releases + Spring Releases + https://repo.spring.io/libs-release + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + + diff --git a/applications/sink/pgcopy-sink/src/main/java/org/springframework/cloud/stream/app/pgcopy/sink/DefaultInitializationScriptResource.java b/applications/sink/pgcopy-sink/src/main/java/org/springframework/cloud/stream/app/pgcopy/sink/DefaultInitializationScriptResource.java new file mode 100644 index 00000000..6c736241 --- /dev/null +++ b/applications/sink/pgcopy-sink/src/main/java/org/springframework/cloud/stream/app/pgcopy/sink/DefaultInitializationScriptResource.java @@ -0,0 +1,59 @@ +/* + * Copyright 2016-2020 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.app.pgcopy.sink; + +import java.nio.charset.Charset; +import java.util.Collection; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.core.io.ByteArrayResource; + +/** + * An in-memory script crafted for dropping-creating the table we're working with. + * All columns are created as VARCHAR(2000). + * + * @author Eric Bottard + * @author Thomas Risberg + */ +public class DefaultInitializationScriptResource extends ByteArrayResource { + + private static final Log logger = LogFactory.getLog(DefaultInitializationScriptResource.class); + + public DefaultInitializationScriptResource(String tableName, Collection columns) { + super(scriptFor(tableName, columns).getBytes(Charset.forName("UTF-8"))); + } + + private static String scriptFor(String tableName, Collection columns) { + StringBuilder result = new StringBuilder("DROP TABLE "); + result.append(tableName).append(";\n\n"); + + result.append("CREATE TABLE ").append(tableName).append('('); + int i = 0; + for (String column : columns) { + if (i++ > 0) { + result.append(", "); + } + result.append(column).append(" VARCHAR(2000)"); + } + result.append(");\n"); + logger.debug(String.format("Generated the following initializing script for table %s:\n%s", tableName, + result.toString())); + return result.toString(); + } +} diff --git a/applications/sink/pgcopy-sink/src/main/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkConfiguration.java b/applications/sink/pgcopy-sink/src/main/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkConfiguration.java new file mode 100644 index 00000000..bb2d0ab8 --- /dev/null +++ b/applications/sink/pgcopy-sink/src/main/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkConfiguration.java @@ -0,0 +1,332 @@ +/* + * Copyright 2016-2019 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.app.pgcopy.sink; + +import java.sql.Connection; +import java.sql.SQLException; +import java.util.Collection; +import java.util.Collections; + +import javax.annotation.PreDestroy; +import javax.sql.DataSource; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.postgresql.copy.CopyIn; +import org.postgresql.copy.CopyManager; +import org.postgresql.core.BaseConnection; + +import org.springframework.beans.factory.FactoryBean; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binding.InputBindingLifecycle; +import org.springframework.cloud.stream.messaging.Sink; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Primary; +import org.springframework.core.io.ResourceLoader; +import org.springframework.dao.DataAccessException; +import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor; +import org.springframework.integration.aggregator.ExpressionEvaluatingCorrelationStrategy; +import org.springframework.integration.aggregator.MessageCountReleaseStrategy; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.config.AggregatorFactoryBean; +import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.store.MessageGroupStoreReaper; +import org.springframework.integration.store.SimpleMessageStore; +import org.springframework.jdbc.core.ConnectionCallback; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.init.DataSourceInitializer; +import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; +import org.springframework.scheduling.annotation.EnableScheduling; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.TransactionCallback; +import org.springframework.transaction.support.TransactionTemplate; +import org.springframework.util.StringUtils; + +/** + * Configuration class for the PostgreSQL CopyManager. + * + * @author Thomas Risberg + * @author Janne Valkealahti + */ +@Configuration +@EnableScheduling +@EnableBinding(Sink.class) +@EnableConfigurationProperties(PgcopySinkProperties.class) +public class PgcopySinkConfiguration { + + private static final Log logger = LogFactory.getLog(PgcopySinkConfiguration.class); + + @Autowired + private PgcopySinkProperties properties; + + @Bean + public MessageChannel toSink() { + return new DirectChannel(); + } + + @Bean + @Primary + @ServiceActivator(inputChannel = Sink.INPUT) + FactoryBean aggregatorFactoryBean(MessageChannel toSink, MessageGroupStore messageGroupStore) { + AggregatorFactoryBean aggregatorFactoryBean = new AggregatorFactoryBean(); + aggregatorFactoryBean.setCorrelationStrategy( + new ExpressionEvaluatingCorrelationStrategy("payload.getClass().name")); + aggregatorFactoryBean.setReleaseStrategy(new MessageCountReleaseStrategy(properties.getBatchSize())); + aggregatorFactoryBean.setMessageStore(messageGroupStore); + aggregatorFactoryBean.setProcessorBean(new DefaultAggregatingMessageGroupProcessor()); + aggregatorFactoryBean.setExpireGroupsUponCompletion(true); + aggregatorFactoryBean.setSendPartialResultOnExpiry(true); + aggregatorFactoryBean.setOutputChannel(toSink); + return aggregatorFactoryBean; + } + + @Bean + @ServiceActivator(inputChannel = "toSink") + public MessageHandler datasetSinkMessageHandler(final JdbcTemplate jdbcTemplate, + final PlatformTransactionManager platformTransactionManager) { + + final TransactionTemplate txTemplate = new TransactionTemplate(platformTransactionManager); + + if (StringUtils.hasText(properties.getErrorTable())) { + verifyErrorTable(jdbcTemplate, txTemplate); + } + + StringBuilder columns = new StringBuilder(); + for (String col : properties.getColumns()) { + if (columns.length() > 0) { + columns.append(","); + } + columns.append(col); + } + // the copy command + final StringBuilder sql = new StringBuilder("COPY " + properties.getTableName()); + if (columns.length() > 0) { + sql.append(" (" + columns + ")"); + } + sql.append(" FROM STDIN"); + + StringBuilder options = new StringBuilder(); + if (properties.getFormat() == PgcopySinkProperties.Format.CSV) { + options.append("CSV"); + } + if (properties.getDelimiter() != null) { + options.append(escapedOptionCharacterValue(options.length(), "DELIMITER", properties.getDelimiter())); + } + if (properties.getNullString() != null) { + options.append((options.length() > 0 ? " " : "") + "NULL '" + properties.getNullString() + "'"); + } + if (properties.getQuote() != null) { + options.append(quotedOptionCharacterValue(options.length(), "QUOTE", properties.getQuote())); + } + if (properties.getEscape() != null) { + options.append(quotedOptionCharacterValue(options.length(), "ESCAPE", properties.getEscape())); + } + if (options.length() > 0) { + sql.append(" WITH " + options.toString()); + } + + return new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + Object payload = message.getPayload(); + if (payload instanceof Collection) { + final Collection payloads = (Collection) payload; + if (logger.isDebugEnabled()) { + logger.debug("Executing batch of size " + payloads.size() + " for " + sql); + } + try { + long rows = doCopy(payloads, txTemplate); + if (logger.isDebugEnabled()) { + logger.debug("Wrote " + rows + " rows"); + } + } + catch (DataAccessException e) { + logger.error("Error while copying batch of data: " + e.getMessage()); + logger.error("Switching to single row copy for current batch"); + long rows = 0; + for (Object singlePayload : payloads) { + try { + rows = rows + doCopy(Collections.singletonList(singlePayload), txTemplate); + } + catch (DataAccessException e2) { + logger.error("Copy for single row caused error: " + e2.getMessage()); + logger.error("Bad Data: \n" + singlePayload); + if (StringUtils.hasText(properties.getErrorTable())) { + writeError(e2, singlePayload); + } + } + } + if (logger.isDebugEnabled()) { + logger.debug("Re-tried batch and wrote " + rows + " rows"); + } + } + } + else { + throw new IllegalStateException("Expected a collection of strings but received " + + message.getPayload().getClass().getName()); + } + } + + private void writeError(final DataAccessException exception, final Object payload) { + final String message; + if (exception.getCause() != null) { + message = exception.getCause().getMessage(); + } + else { + message = exception.getMessage(); + } + try { + txTemplate.execute(new TransactionCallback() { + @Override + public Long doInTransaction(TransactionStatus transactionStatus) { + jdbcTemplate.update( + "insert into " + properties.getErrorTable() + " (table_name, error_message, payload) values (?, ?, ?)", + new Object[]{properties.getTableName(), message, payload}); + return null; + } + }); + } + catch (DataAccessException e) { + logger.error("Writing to error table failed: " + e.getMessage()); + } + } + + private long doCopy(final Collection payloads, TransactionTemplate txTemplate) { + Long rows = txTemplate.execute(transactionStatus -> jdbcTemplate.execute( + new ConnectionCallback() { + @Override + public Long doInConnection(Connection connection) throws SQLException, DataAccessException { + CopyManager cm = connection.unwrap(BaseConnection.class).getCopyAPI(); + CopyIn ci = cm.copyIn(sql.toString()); + for (Object payloadData : payloads) { + String textPayload = (payloadData instanceof byte[]) ? + new String((byte[]) payloadData) : (String) payloadData; + byte[] data = (textPayload + "\n").getBytes(); + ci.writeToCopy(data, 0, data.length); + } + return Long.valueOf(ci.endCopy()); + } + } + )); + return rows; + } + }; + } + + @ConditionalOnProperty("pgcopy.initialize") + @Bean + public DataSourceInitializer nonBootDataSourceInitializer(DataSource dataSource, ResourceLoader resourceLoader) { + DataSourceInitializer dataSourceInitializer = new DataSourceInitializer(); + dataSourceInitializer.setDataSource(dataSource); + ResourceDatabasePopulator databasePopulator = new ResourceDatabasePopulator(); + databasePopulator.setIgnoreFailedDrops(true); + dataSourceInitializer.setDatabasePopulator(databasePopulator); + if ("true".equals(properties.getInitialize())) { + databasePopulator.addScript(new DefaultInitializationScriptResource(properties.getTableName(), + properties.getColumns())); + } + else { + databasePopulator.addScript(resourceLoader.getResource(properties.getInitialize())); + } + return dataSourceInitializer; + } + + @Bean + MessageGroupStore messageGroupStore() { + SimpleMessageStore messageGroupStore = new SimpleMessageStore(); + messageGroupStore.setTimeoutOnIdle(true); + messageGroupStore.setCopyOnGet(false); + return messageGroupStore; + } + + @Bean + MessageGroupStoreReaper messageGroupStoreReaper(MessageGroupStore messageStore, + InputBindingLifecycle inputBindingLifecycle) { + MessageGroupStoreReaper messageGroupStoreReaper = new MessageGroupStoreReaper(messageStore); + messageGroupStoreReaper.setPhase(inputBindingLifecycle.getPhase() - 1); + messageGroupStoreReaper.setTimeout(properties.getIdleTimeout()); + messageGroupStoreReaper.setAutoStartup(true); + messageGroupStoreReaper.setExpireOnDestroy(true); + return messageGroupStoreReaper; + } + + @Bean + ReaperTask reaperTask() { + return new ReaperTask(); + } + + @Bean + public JdbcTemplate jdbcTemplate(DataSource dataSource) { + JdbcTemplate jt = new JdbcTemplate(dataSource); + return jt; + } + + private String quotedOptionCharacterValue(int length, String option, char value) { + return (length > 0 ? " " : "") + option + " '" + (value == '\'' ? "''" : value) + "'"; + } + + private String escapedOptionCharacterValue(int length, String option, String value) { + return (length > 0 ? " " : "") + option + " " + (value.startsWith("\\") ? "E'" + value : "'" + value) + "'"; + } + + private void verifyErrorTable(final JdbcTemplate jdbcTemplate, final TransactionTemplate txTemplate) { + try { + txTemplate.execute(new TransactionCallback() { + @Override + public Long doInTransaction(TransactionStatus transactionStatus) { + jdbcTemplate.update( + "insert into " + properties.getErrorTable() + " (table_name, error_message, payload) values (?, ?, ?)", + properties.getErrorTable(), "message", "payload"); + transactionStatus.setRollbackOnly(); + return null; + } + }); + } + catch (DataAccessException e) { + throw new IllegalStateException("Invalid error table specified", e); + } + } + + public static class ReaperTask { + + @Autowired + MessageGroupStoreReaper messageGroupStoreReaper; + + @Scheduled(fixedRate = 1000) + public void reap() { + messageGroupStoreReaper.run(); + } + + @PreDestroy + public void beforeDestroy() { + reap(); + } + + } +} diff --git a/applications/sink/pgcopy-sink/src/main/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkProperties.java b/applications/sink/pgcopy-sink/src/main/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkProperties.java new file mode 100644 index 00000000..35a3a9c7 --- /dev/null +++ b/applications/sink/pgcopy-sink/src/main/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkProperties.java @@ -0,0 +1,204 @@ +/* + * Copyright 2016-2017 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.app.pgcopy.sink; + +import java.util.Collections; +import java.util.List; + +import javax.validation.constraints.NotNull; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.validation.annotation.Validated; + +/** + * Used to configure the pgcopy sink module options that are related to writing using the PostgreSQL CopyManager API. + * + * @author Thomas Risberg + */ +@SuppressWarnings("unused") +@ConfigurationProperties("pgcopy") +@Validated +public class PgcopySinkProperties { + + /** + * The name of the table to write into. + */ + @NotNull + private String tableName; + + /** + * The names of the columns that shall receive data. + * Also used at initialization time to issue the DDL. + */ + private List columns = Collections.singletonList("payload"); + + /** + * Threshold in number of messages when data will be flushed to database table. + */ + private int batchSize = 10000; + + /** + * Idle timeout in milliseconds when data is automatically flushed to database table. + */ + private long idleTimeout = -1L; + + /** + * 'true', 'false' or the location of a custom initialization script for the table. + */ + private String initialize = "false"; + + /** + * Format to use for the copy command. + */ + private Format format = Format.TEXT; + + /** + * Specifies the string that represents a null value. The default is \N (backslash-N) in text format, and an + * unquoted empty string in CSV format. + */ + private String nullString; + + /** + * Specifies the character that separates columns within each row (line) of the file. The default is a tab character + * in text format, a comma in CSV format. This must be a single one-byte character. Using an escaped value like '\t' + * is allowed. + */ + private String delimiter; + + /** + * Specifies the quoting character to be used when a data value is quoted. The default is double-quote. This must + * be a single one-byte character. This option is allowed only when using CSV format. + */ + private Character quote; + + /** + * Specifies the character that should appear before a data character that matches the QUOTE value. The default is + * the same as the QUOTE value (so that the quoting character is doubled if it appears in the data). This must be + * a single one-byte character. This option is allowed only when using CSV format. + */ + private Character escape; + + /** + * The name of the error table used for writing rows causing errors. The error table should have three columns + * named "table_name", "error_message" and "payload" large enough to hold potential data values. + * You can use the following DDL to create this table: + * 'CREATE TABLE ERRORS (TABLE_NAME VARCHAR(255), ERROR_MESSAGE TEXT,PAYLOAD TEXT)' + */ + private String errorTable; + + + public String getTableName() { + return tableName; + } + + public void setTableName(String tableName) { + this.tableName = tableName; + } + + public List getColumns() { + return columns; + } + + public void setColumns(List columns) { + this.columns = columns; + } + + public int getBatchSize() { + return batchSize; + } + + public void setBatchSize(int batchSize) { + this.batchSize = batchSize; + } + + public long getIdleTimeout() { + return idleTimeout; + } + + public void setIdleTimeout(long idleTimeout) { + this.idleTimeout = idleTimeout; + } + + public String getInitialize() { + return initialize; + } + + public void setInitialize(String initialize) { + this.initialize = initialize; + } + + public Format getFormat() { + return format; + } + + public void setFormat(Format format) { + this.format = format; + } + + public String getNullString() { + return nullString; + } + + public void setNullString(String nullString) { + this.nullString = nullString; + } + + public String getDelimiter() { + return delimiter; + } + + public void setDelimiter(String delimiter) { + this.delimiter = delimiter; + } + + public Character getQuote() { + return quote; + } + + public void setQuote(Character quote) { + this.quote = quote; + } + + public Character getEscape() { + return escape; + } + + public void setEscape(Character escape) { + this.escape = escape; + } + + public String getErrorTable() { + return errorTable; + } + + public void setErrorTable(String errorTable) { + this.errorTable = errorTable; + } + + public enum Format { + + /** + * text. + */ + TEXT, + /** + * csv. + */ + CSV + + } +} diff --git a/applications/sink/pgcopy-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/sink/pgcopy-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..f27649d6 --- /dev/null +++ b/applications/sink/pgcopy-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1,6 @@ +configuration-properties.classes=org.springframework.cloud.stream.app.pgcopy.sink.PgcopySinkProperties +configuration-properties.names=\ +spring.datasource.url,\ +spring.datasource.driver-class-name,\ +spring.datasource.username,\ +spring.datasource.password diff --git a/applications/sink/pgcopy-sink/src/main/resources/META-INF/spring-configuration-metadata-whitelist.properties b/applications/sink/pgcopy-sink/src/main/resources/META-INF/spring-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..f27649d6 --- /dev/null +++ b/applications/sink/pgcopy-sink/src/main/resources/META-INF/spring-configuration-metadata-whitelist.properties @@ -0,0 +1,6 @@ +configuration-properties.classes=org.springframework.cloud.stream.app.pgcopy.sink.PgcopySinkProperties +configuration-properties.names=\ +spring.datasource.url,\ +spring.datasource.driver-class-name,\ +spring.datasource.username,\ +spring.datasource.password diff --git a/applications/sink/pgcopy-sink/src/main/resources/META-INF/spring.provides b/applications/sink/pgcopy-sink/src/main/resources/META-INF/spring.provides new file mode 100644 index 00000000..4841cf95 --- /dev/null +++ b/applications/sink/pgcopy-sink/src/main/resources/META-INF/spring.provides @@ -0,0 +1 @@ +provides: spring-cloud-starter-stream-sink-pgcopy \ No newline at end of file diff --git a/applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopyBadErrorTableIntegrationTests.java b/applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopyBadErrorTableIntegrationTests.java new file mode 100644 index 00000000..187951e8 --- /dev/null +++ b/applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopyBadErrorTableIntegrationTests.java @@ -0,0 +1,170 @@ +/* + * Copyright 2016-2019 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.app.pgcopy.sink; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.ClassRule; +import org.junit.Test; +import org.postgresql.util.PSQLException; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.test.util.TestPropertyValues; +import org.springframework.cloud.stream.app.pgcopy.test.PostgresTestSupport; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.core.io.support.PropertiesLoaderUtils; +import org.springframework.dao.DataAccessException; +import org.springframework.jdbc.core.JdbcOperations; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DriverManagerDataSource; + +import static org.hamcrest.Matchers.is; + +/** + * Integration Tests testing bad error table specified for PgcopySink. Only runs if PostgreSQL database is available. + * + * @author Thomas Risberg + * @author Artem Bilan + */ +public class PgcopyBadErrorTableIntegrationTests { + + @ClassRule + public static PostgresTestSupport postgresAvailable = new PostgresTestSupport(); + + private String[] env = { "pgcopy.tableName=names", "pgcopy.columns=id,name,age", "pgcopy.format=CSV" }; + + private String[] jdbc = { }; + + private Properties appProperties = new Properties(); + + @Before + public void setup() { + try { + appProperties = PropertiesLoaderUtils.loadAllProperties("application.properties"); + } + catch (IOException e) { + } + List jdbcProperties = new ArrayList<>(); + for (Object key : appProperties.keySet()) { + jdbcProperties.add(key + "=" + appProperties.get(key)); + } + this.jdbc = jdbcProperties.toArray(new String[0]); + } + + @Test + public void testBadErrorTableName() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + TestPropertyValues.of(this.jdbc) + .and("pgcopy.error-table=missing") + .and(this.env) + .applyTo(context); + context.register(PgcopySinkApplication.class); + try { + context.refresh(); + } + catch (Exception e) { + Throwable ise = null; + Throwable dae = null; + Throwable cause = e; + while (cause.getCause() != null) { + cause = cause.getCause(); + if (cause instanceof IllegalStateException) { + ise = cause; + } + if (cause instanceof DataAccessException) { + dae = cause; + } + } + Assert.assertThat(cause.getClass().getName(), is(PSQLException.class.getName())); + Assert.assertNotNull(ise); + Assert.assertTrue(ise.getMessage().contains("Invalid error table specified")); + Assert.assertNotNull(dae); + Assert.assertTrue(cause.getMessage().contains("relation")); + Assert.assertTrue(cause.getMessage().contains("does not exist")); + } + context.close(); + } + + @Test + public void testBadErrorTableFields() { + DriverManagerDataSource dataSource = new DriverManagerDataSource(); + dataSource.setDriverClassName(appProperties.getProperty("spring.datasource.driver-class-name")); + dataSource.setUrl(appProperties.getProperty("spring.datasource.url")); + dataSource.setUsername(appProperties.getProperty("spring.datasource.username")); + dataSource.setPassword(appProperties.getProperty("spring.datasource.password")); + JdbcOperations jdbcOperations = new JdbcTemplate(dataSource); + try { + jdbcOperations.execute( + "drop table test_errors"); + } + catch (Exception e) { + } + try { + jdbcOperations.execute( + "create table test_errors (table_name varchar(255), error_message text)"); + } + catch (Exception e) { + throw new IllegalStateException("Error creating table", e); + } + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + TestPropertyValues.of(this.jdbc) + .and("pgcopy.error-table=test_errors") + .and(this.env) + .applyTo(context); + context.register(PgcopySinkApplication.class); + try { + context.refresh(); + } + catch (Exception e) { + Throwable ise = null; + Throwable dae = null; + Throwable cause = e; + while (cause.getCause() != null) { + cause = cause.getCause(); + if (cause instanceof IllegalStateException) { + ise = cause; + } + if (cause instanceof DataAccessException) { + dae = cause; + } + } + Assert.assertThat(cause.getClass().getName(), is(PSQLException.class.getName())); + Assert.assertNotNull(ise); + Assert.assertTrue(ise.getMessage().contains("Invalid error table specified")); + Assert.assertNotNull(dae); + Assert.assertTrue(cause.getMessage().contains("column")); + Assert.assertTrue(cause.getMessage().contains("does not exist")); + } + context.close(); + } + + @SpringBootApplication + public static class PgcopySinkApplication { + + public static void main(String[] args) { + SpringApplication.run(PgcopySinkApplication.class, args); + } + + } + +} diff --git a/applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopyErrorTableIntegrationTests.java b/applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopyErrorTableIntegrationTests.java new file mode 100644 index 00000000..1adf148b --- /dev/null +++ b/applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopyErrorTableIntegrationTests.java @@ -0,0 +1,80 @@ +/* + * Copyright 2016-2019 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.app.pgcopy.sink; + +import org.junit.Assert; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.stream.app.pgcopy.test.PostgresTestSupport; +import org.springframework.cloud.stream.messaging.Sink; +import org.springframework.jdbc.core.JdbcOperations; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.hamcrest.Matchers.is; + +/** + * Integration Tests for PgcopySink with error table. Only runs if PostgreSQL database is available. + * + * @author Thomas Risberg + * @author Janne Valkealahti + */ +@RunWith(SpringRunner.class) +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, + classes = PgcopyErrorTableIntegrationTests.PgcopySinkApplication.class) +@TestPropertySource(properties = { "pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true", + "pgcopy.columns=id,name,age", "pgcopy.format=CSV", "pgcopy.error-table=test_errors", + "spring.datasource.initialization-mode=always", "spring.datasource.schema=classpath:error-table-ddl.sql", + "spring.datasource.continue-on-error=true" }) +@DirtiesContext(classMode = DirtiesContext.ClassMode.BEFORE_EACH_TEST_METHOD) +public class PgcopyErrorTableIntegrationTests { + + @ClassRule + public static PostgresTestSupport postgresAvailable = new PostgresTestSupport(); + + @Autowired + protected Sink channels; + + @Autowired + protected JdbcOperations jdbcOperations; + + @Test + public void testCopyCSV() { + channels.input().send(MessageBuilder.withPayload("123,Nisse,25").build()); + channels.input().send(MessageBuilder.withPayload("GARBAGE").build()); + channels.input().send(MessageBuilder.withPayload("125,Bubba,22").build()); + int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class); + int errors = jdbcOperations.queryForObject("select count(*) from test_errors", Integer.class); + Assert.assertThat(result, is(2)); + Assert.assertThat(errors, is(1)); + } + + @SpringBootApplication + public static class PgcopySinkApplication { + public static void main(String[] args) { + SpringApplication.run(PgcopySinkApplication.class, args); + } + } +} diff --git a/applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkIntegrationTests.java b/applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkIntegrationTests.java new file mode 100644 index 00000000..ca2258f9 --- /dev/null +++ b/applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkIntegrationTests.java @@ -0,0 +1,197 @@ +/* + * Copyright 2016-2019 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.app.pgcopy.sink; + +import org.junit.Assert; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.stream.app.pgcopy.test.PostgresTestSupport; +import org.springframework.cloud.stream.messaging.Sink; +import org.springframework.jdbc.core.JdbcOperations; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.TestPropertySource; +import org.springframework.test.context.junit4.SpringRunner; + +import static org.hamcrest.Matchers.is; + +/** + * Integration Tests for PgcopySink. Only runs if PostgreSQL database is available. + * + * @author Thomas Risberg + */ +@RunWith(SpringRunner.class) +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, + classes = PgcopySinkIntegrationTests.PgcopySinkApplication.class) +@DirtiesContext +public abstract class PgcopySinkIntegrationTests { + + @ClassRule + public static PostgresTestSupport postgresAvailable = new PostgresTestSupport(); + + @Autowired + protected Sink channels; + + @Autowired + protected JdbcOperations jdbcOperations; + + @TestPropertySource(properties = {"pgcopy.table-name=test", "pgcopy.batch-size=1", "pgcopy.initialize=true"}) + public static class BasicPayloadCopyTests extends PgcopySinkIntegrationTests { + + @Test + public void testBasicCopy() { + String sent = "hello42"; + channels.input().send(MessageBuilder.withPayload(sent).build()); + String result = jdbcOperations.queryForObject("select payload from test", String.class); + Assert.assertThat(result, is("hello42")); + } + } + + @TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=4", "pgcopy.initialize=true", + "pgcopy.columns=id,name,age"}) + public static class PgcopyTextTests extends PgcopySinkIntegrationTests { + + @Test + public void testCopyText() { + channels.input().send(MessageBuilder.withPayload("123\tNisse\t25").build()); + channels.input().send(MessageBuilder.withPayload("124\tAnna\t21").build()); + channels.input().send(MessageBuilder.withPayload("125\tBubba\t22").build()); + channels.input().send(MessageBuilder.withPayload("126\tPelle\t32").build()); + int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class); + Assert.assertThat(result, is(4)); + } + } + + @TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true", + "pgcopy.columns=id,name,age", "pgcopy.format=CSV"}) + public static class PgcopyCSVTests extends PgcopySinkIntegrationTests { + + @Test + public void testCopyCSV() { + channels.input().send(MessageBuilder.withPayload("123,\"Nisse\",25").build()); + channels.input().send(MessageBuilder.withPayload("124,\"Anna\",21").build()); + channels.input().send(MessageBuilder.withPayload("125,\"Bubba\",22").build()); + int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class); + Assert.assertThat(result, is(3)); + } + } + + @TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true", + "pgcopy.columns=id,name,age", "pgcopy.format=CSV"}) + public static class PgcopyNullTests extends PgcopySinkIntegrationTests { + + @Test + public void testCopyCSV() { + channels.input().send(MessageBuilder.withPayload("123,\"Nisse\",25").build()); + channels.input().send(MessageBuilder.withPayload("124,,21").build()); + channels.input().send(MessageBuilder.withPayload("125,\"Bubba\",22").build()); + int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class); + int nulls = jdbcOperations.queryForObject("select count(*) from names where name is null", Integer.class); + Assert.assertThat(result, is(3)); + Assert.assertThat(nulls, is(1)); + } + } + + @TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true", + "pgcopy.columns=id,name,age", "pgcopy.format=CSV", "pgcopy.null-string=null"}) + public static class PgcopyNullStringTests extends PgcopySinkIntegrationTests { + + @Test + public void testCopyCSV() { + channels.input().send(MessageBuilder.withPayload("123,\"Nisse\",25").build()); + channels.input().send(MessageBuilder.withPayload("124,null,21").build()); + channels.input().send(MessageBuilder.withPayload("125,\"Bubba\",22").build()); + int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class); + int nulls = jdbcOperations.queryForObject("select count(*) from names where name is null", Integer.class); + Assert.assertThat(result, is(3)); + Assert.assertThat(nulls, is(1)); + } + } + + @TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true", + "pgcopy.columns=id,name,age", "pgcopy.format=CSV", "pgcopy.delimiter=|"}) + public static class PgcopyDelimiterTests extends PgcopySinkIntegrationTests { + + @Test + public void testCopyCSV() { + channels.input().send(MessageBuilder.withPayload("123|\"Nisse\"|25").build()); + channels.input().send(MessageBuilder.withPayload("124|\"Anna\"|21").build()); + channels.input().send(MessageBuilder.withPayload("125|\"Bubba\"|22").build()); + int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class); + Assert.assertThat(result, is(3)); + } + } + + @TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true", + "pgcopy.columns=id,name,age", "pgcopy.format=CSV", "pgcopy.delimiter=\\t"}) + public static class PgcopyEscapedDelimiterTests extends PgcopySinkIntegrationTests { + + @Test + public void testCopyCSV() { + channels.input().send(MessageBuilder.withPayload("123\t\"Nisse\"\t25").build()); + channels.input().send(MessageBuilder.withPayload("124\t\"Anna\"\t21").build()); + channels.input().send(MessageBuilder.withPayload("125\t\"Bubba\"\t22").build()); + int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class); + Assert.assertThat(result, is(3)); + } + } + + @TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true", + "pgcopy.columns=id,name,age", "pgcopy.format=CSV", "pgcopy.quote='"}) + public static class PgcopyQuoteTests extends PgcopySinkIntegrationTests { + + @Test + public void testCopyCSV() { + channels.input().send(MessageBuilder.withPayload("123,Nisse,25").build()); + channels.input().send(MessageBuilder.withPayload("124,'Anna',21").build()); + channels.input().send(MessageBuilder.withPayload("125,Bubba,22").build()); + int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class); + int quoted = jdbcOperations.queryForObject("select count(*) from names where name = 'Anna'", Integer.class); + Assert.assertThat(result, is(3)); + Assert.assertThat(quoted, is(1)); + } + } + + @TestPropertySource(properties = {"pgcopy.tableName=names", "pgcopy.batch-size=3", "pgcopy.initialize=true", + "pgcopy.columns=id,name,age", "pgcopy.format=CSV", "pgcopy.escape=\\\\"}) + public static class PgcopyEscapeTests extends PgcopySinkIntegrationTests { + + @Test + public void testCopyCSV() { + channels.input().send(MessageBuilder.withPayload("123,Nisse,25").build()); + channels.input().send(MessageBuilder.withPayload("124,\"Anna\\\"\",21").build()); + channels.input().send(MessageBuilder.withPayload("125,Bubba,22").build()); + int result = jdbcOperations.queryForObject("select count(*) from names", Integer.class); + int quoted = jdbcOperations.queryForObject("select count(*) from names where name = 'Anna\"'", Integer.class); + Assert.assertThat(result, is(3)); + Assert.assertThat(quoted, is(1)); + } + } + + @SpringBootApplication + public static class PgcopySinkApplication { + public static void main(String[] args) { + SpringApplication.run(PgcopySinkApplication.class, args); + } + } +} diff --git a/applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkPropertiesTests.java b/applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkPropertiesTests.java new file mode 100644 index 00000000..8bed14e2 --- /dev/null +++ b/applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/sink/PgcopySinkPropertiesTests.java @@ -0,0 +1,153 @@ +/* + * Copyright 2017-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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.app.pgcopy.sink; + +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; + +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.test.util.TestPropertyValues; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Configuration; + +import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.MatcherAssert.assertThat; + +/** + * @author Thomas Risberg + * @author Artem Bilan + */ +public class PgcopySinkPropertiesTests { + + private AnnotationConfigApplicationContext context; + + @Rule + public ExpectedException thrown = ExpectedException.none(); + + @Before + public void setUp() { + this.context = new AnnotationConfigApplicationContext(); + } + + @After + public void tearDown() { + this.context.close(); + } + + @Test + public void tableNameIsRequired() { + this.thrown.expect(BeanCreationException.class); + this.thrown.expectMessage("Failed to bind properties under 'pgcopy' to org.springframework.cloud.stream.app.pgcopy.sink.PgcopySinkProperties"); + this.context.register(Conf.class); + this.context.refresh(); + } + + @Test + public void tableNameCanBeCustomized() { + String table = "TEST_DATA"; + TestPropertyValues.of("pgcopy.table-name:" + table) + .applyTo(this.context); + this.context.register(Conf.class); + this.context.refresh(); + PgcopySinkProperties properties = this.context.getBean(PgcopySinkProperties.class); + assertThat(properties.getTableName(), equalTo(table)); + } + + @Test + public void formatDefaultsToText() { + TestPropertyValues.of("pgcopy.table-name: test") + .applyTo(this.context); + this.context.register(Conf.class); + this.context.refresh(); + PgcopySinkProperties properties = this.context.getBean(PgcopySinkProperties.class); + assertThat(properties.getFormat().toString(), equalTo("TEXT")); + } + + @Test + public void formatCanBeCustomized() { + String format = "CSV"; + TestPropertyValues.of("pgcopy.table-name: test", "pgcopy.format:" + format) + .applyTo(this.context); + this.context.register(Conf.class); + this.context.refresh(); + PgcopySinkProperties properties = this.context.getBean(PgcopySinkProperties.class); + assertThat(properties.getFormat().toString(), equalTo(format)); + } + + @Test + public void nullCanBeCustomized() { + String nullString = "@#$"; + TestPropertyValues.of("pgcopy.table-name: test", + "pgcopy.format: CSV", + "pgcopy.null-string: " + nullString) + .applyTo(this.context); + this.context.register(Conf.class); + this.context.refresh(); + PgcopySinkProperties properties = this.context.getBean(PgcopySinkProperties.class); + assertThat(properties.getNullString(), equalTo(nullString)); + } + + @Test + public void delimiterCanBeCustomized() { + String delimiter = "|"; + TestPropertyValues.of("pgcopy.table-name: test", + "pgcopy.format: CSV", + "pgcopy.delimiter: " + delimiter) + .applyTo(this.context); + this.context.register(Conf.class); + this.context.refresh(); + PgcopySinkProperties properties = this.context.getBean(PgcopySinkProperties.class); + assertThat(properties.getDelimiter(), equalTo(delimiter)); + } + + @Test + public void quoteCanBeCustomized() { + String quote = "'"; + TestPropertyValues.of("pgcopy.table-name: test", + "pgcopy.format: CSV", + "pgcopy.quote: " + quote) + .applyTo(this.context); + this.context.register(Conf.class); + this.context.refresh(); + PgcopySinkProperties properties = this.context.getBean(PgcopySinkProperties.class); + assertThat(String.valueOf(properties.getQuote()), equalTo(quote)); + } + + @Test + public void escapeCanBeCustomized() { + String escape = "~"; + TestPropertyValues.of("pgcopy.table-name: test", + "pgcopy.format: CSV", + "pgcopy.escape: " + escape) + .applyTo(this.context); + this.context.register(Conf.class); + this.context.refresh(); + PgcopySinkProperties properties = this.context.getBean(PgcopySinkProperties.class); + assertThat(String.valueOf(properties.getEscape()), equalTo(escape)); + } + + @Configuration + @EnableConfigurationProperties(PgcopySinkProperties.class) + static class Conf { + + } + +} diff --git a/applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/test/PostgresTestSupport.java b/applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/test/PostgresTestSupport.java new file mode 100644 index 00000000..f3eab03e --- /dev/null +++ b/applications/sink/pgcopy-sink/src/test/java/org/springframework/cloud/stream/app/pgcopy/test/PostgresTestSupport.java @@ -0,0 +1,66 @@ +/* + * Copyright 2017-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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.app.pgcopy.test; + +import java.sql.Connection; + +import javax.sql.DataSource; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.test.junit.AbstractExternalResourceTestSupport; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.datasource.DataSourceUtils; + +/** + * JUnit {@link org.junit.Rule} that detects the fact that a PostgreSQL server is running on localhost. + * + * @author Thomas Risberg + * @author Artem Bilan + */ +public class PostgresTestSupport extends AbstractExternalResourceTestSupport { + + private ConfigurableApplicationContext context; + + public PostgresTestSupport() { + super("POSTGRES"); + } + + @Override + protected void cleanupResource() { + context.close(); + } + + @Override + protected void obtainResource() { + context = new SpringApplicationBuilder(Config.class) + .web(WebApplicationType.NONE) + .run(); + DataSource dataSource = context.getBean(DataSource.class); + Connection con = DataSourceUtils.getConnection(dataSource); + DataSourceUtils.releaseConnection(con, dataSource); + } + + @Configuration + @EnableAutoConfiguration + public static class Config { + + } + +} diff --git a/applications/sink/pgcopy-sink/src/test/resources/application.properties b/applications/sink/pgcopy-sink/src/test/resources/application.properties new file mode 100644 index 00000000..e03d53d1 --- /dev/null +++ b/applications/sink/pgcopy-sink/src/test/resources/application.properties @@ -0,0 +1,4 @@ +spring.datasource.url=jdbc:postgresql://127.0.0.1:5432/test +spring.datasource.username=postgres +spring.datasource.password=spring +spring.datasource.driver-class-name=org.postgresql.Driver diff --git a/applications/sink/pgcopy-sink/src/test/resources/error-table-ddl.sql b/applications/sink/pgcopy-sink/src/test/resources/error-table-ddl.sql new file mode 100644 index 00000000..45c207ff --- /dev/null +++ b/applications/sink/pgcopy-sink/src/test/resources/error-table-ddl.sql @@ -0,0 +1,2 @@ +drop table test_errors; +create table test_errors (table_name varchar(255), error_message text, payload text); diff --git a/applications/sink/pom.xml b/applications/sink/pom.xml index 29868e94..515b7327 100644 --- a/applications/sink/pom.xml +++ b/applications/sink/pom.xml @@ -31,5 +31,6 @@ twitter-update-sink twitter-message-sink wavefront-sink + pgcopy-sink