You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by zh...@apache.org on 2022/12/16 02:17:56 UTC
[shardingsphere] branch master updated: Remove spring-jdbc dependency at pipeline e2e test (#22890)
This is an automated email from the ASF dual-hosted git repository.
zhonghongsheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new 287c7b71853 Remove spring-jdbc dependency at pipeline e2e test (#22890)
287c7b71853 is described below
commit 287c7b718532c374072819b2594f45e79c27fda6
Author: Xinze Guo <10...@users.noreply.github.com>
AuthorDate: Fri Dec 16 10:17:50 2022 +0800
Remove spring-jdbc dependency at pipeline e2e test (#22890)
---
test/e2e/pipeline/pom.xml | 9 ---
.../pipeline/cases/base/PipelineBaseE2EIT.java | 10 +--
.../general/MySQLMigrationGeneralE2EIT.java | 13 ++--
.../general/PostgreSQLMigrationGeneralE2EIT.java | 17 ++---
.../pipeline/cases/task/MySQLIncrementTask.java | 22 ++++--
.../cases/task/PostgreSQLIncrementTask.java | 25 +++---
.../data/pipeline/util/DataSourceExecuteUtil.java | 89 ++++++++++++++++++++++
7 files changed, 136 insertions(+), 49 deletions(-)
diff --git a/test/e2e/pipeline/pom.xml b/test/e2e/pipeline/pom.xml
index 18535d8c726..cb6e5c624ba 100644
--- a/test/e2e/pipeline/pom.xml
+++ b/test/e2e/pipeline/pom.xml
@@ -29,8 +29,6 @@
<properties>
<maven.deploy.skip>true</maven.deploy.skip>
-
- <springframework.version>4.3.6.RELEASE</springframework.version>
</properties>
<dependencies>
@@ -62,13 +60,6 @@
<type>test-jar</type>
</dependency>
- <dependency>
- <groupId>org.springframework</groupId>
- <artifactId>spring-jdbc</artifactId>
- <version>${springframework.version}</version>
- <scope>test</scope>
- </dependency>
-
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
diff --git a/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/base/PipelineBaseE2EIT.java b/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/base/PipelineBaseE2EIT.java
index f33d7b6ecef..3e3d11ed76e 100644
--- a/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/base/PipelineBaseE2EIT.java
+++ b/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/base/PipelineBaseE2EIT.java
@@ -20,7 +20,6 @@ package org.apache.shardingsphere.test.e2e.data.pipeline.cases.base;
import com.google.common.base.Strings;
import lombok.AccessLevel;
import lombok.Getter;
-import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.data.pipeline.api.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.core.util.ThreadUtil;
@@ -106,7 +105,6 @@ public abstract class PipelineBaseE2EIT {
private DataSource proxyDataSource;
- @Setter
private Thread increaseTaskThread;
public PipelineBaseE2EIT(final PipelineTestParameter testParam) {
@@ -271,14 +269,14 @@ public abstract class PipelineBaseE2EIT {
}
protected void startIncrementTask(final BaseIncrementTask baseIncrementTask) {
- setIncreaseTaskThread(new Thread(baseIncrementTask));
- getIncreaseTaskThread().start();
+ increaseTaskThread = new Thread(baseIncrementTask);
+ increaseTaskThread.start();
}
// TODO use DAO to query via DistSQL
protected List<Map<String, Object>> waitIncrementTaskFinished(final String distSQL) throws InterruptedException {
- if (null != getIncreaseTaskThread()) {
- TimeUnit.SECONDS.timedJoin(getIncreaseTaskThread(), 60);
+ if (null != increaseTaskThread) {
+ TimeUnit.SECONDS.timedJoin(increaseTaskThread, 30);
}
for (int i = 0; i < 10; i++) {
List<Map<String, Object>> listJobStatus = queryForListWithLog(distSQL);
diff --git a/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/general/MySQLMigrationGeneralE2EIT.java b/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/general/MySQLMigrationGeneralE2EIT.java
index ce08972e824..35924dc38b6 100644
--- a/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/general/MySQLMigrationGeneralE2EIT.java
+++ b/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/general/MySQLMigrationGeneralE2EIT.java
@@ -20,18 +20,18 @@ package org.apache.shardingsphere.test.e2e.data.pipeline.cases.migration.general
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.shardingsphere.infra.database.type.dialect.MySQLDatabaseType;
+import org.apache.shardingsphere.sharding.algorithm.keygen.SnowflakeKeyGenerateAlgorithm;
+import org.apache.shardingsphere.test.e2e.data.pipeline.cases.base.PipelineBaseE2EIT;
import org.apache.shardingsphere.test.e2e.data.pipeline.cases.migration.AbstractMigrationE2EIT;
import org.apache.shardingsphere.test.e2e.data.pipeline.cases.task.MySQLIncrementTask;
import org.apache.shardingsphere.test.e2e.data.pipeline.env.enums.PipelineEnvTypeEnum;
import org.apache.shardingsphere.test.e2e.data.pipeline.framework.helper.PipelineCaseHelper;
import org.apache.shardingsphere.test.e2e.data.pipeline.framework.param.PipelineTestParameter;
-import org.apache.shardingsphere.sharding.algorithm.keygen.SnowflakeKeyGenerateAlgorithm;
-import org.apache.shardingsphere.test.e2e.data.pipeline.cases.base.PipelineBaseE2EIT;
+import org.apache.shardingsphere.test.e2e.data.pipeline.util.DataSourceExecuteUtil;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
-import org.springframework.jdbc.core.JdbcTemplate;
import java.sql.SQLException;
import java.time.LocalDateTime;
@@ -84,17 +84,16 @@ public final class MySQLMigrationGeneralE2EIT extends AbstractMigrationE2EIT {
createTargetOrderTableRule();
createTargetOrderTableEncryptRule();
createTargetOrderItemTableRule();
- JdbcTemplate jdbcTemplate = new JdbcTemplate(getSourceDataSource());
Pair<List<Object[]>, List<Object[]>> dataPair = PipelineCaseHelper.generateFullInsertData(testParam.getDatabaseType(), 3000);
log.info("init data begin: {}", LocalDateTime.now());
- jdbcTemplate.batchUpdate(getExtraSQLCommand().getFullInsertOrder(getSourceTableOrderName()), dataPair.getLeft());
- jdbcTemplate.batchUpdate(getExtraSQLCommand().getFullInsertOrderItem(), dataPair.getRight());
+ DataSourceExecuteUtil.execute(getSourceDataSource(), getExtraSQLCommand().getFullInsertOrder(getSourceTableOrderName()), dataPair.getLeft());
+ DataSourceExecuteUtil.execute(getSourceDataSource(), getExtraSQLCommand().getFullInsertOrderItem(), dataPair.getRight());
log.info("init data end: {}", LocalDateTime.now());
startMigration(getSourceTableOrderName(), getTargetTableOrderName());
startMigration("t_order_item", "t_order_item");
String orderJobId = getJobIdByTableName(getSourceTableOrderName());
waitJobPrepareSuccess(String.format("SHOW MIGRATION STATUS '%s'", orderJobId));
- startIncrementTask(new MySQLIncrementTask(jdbcTemplate, getSourceTableOrderName(), new SnowflakeKeyGenerateAlgorithm(), 30));
+ startIncrementTask(new MySQLIncrementTask(getSourceDataSource(), getSourceTableOrderName(), new SnowflakeKeyGenerateAlgorithm(), 30));
assertMigrationSuccessById(orderJobId, "DATA_MATCH");
String orderItemJobId = getJobIdByTableName("t_order_item");
assertMigrationSuccessById(orderItemJobId, "DATA_MATCH");
diff --git a/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/general/PostgreSQLMigrationGeneralE2EIT.java b/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/general/PostgreSQLMigrationGeneralE2EIT.java
index b7e319fde5b..5cbb693c1f6 100644
--- a/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/general/PostgreSQLMigrationGeneralE2EIT.java
+++ b/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/migration/general/PostgreSQLMigrationGeneralE2EIT.java
@@ -21,18 +21,18 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.shardingsphere.infra.database.type.dialect.OpenGaussDatabaseType;
import org.apache.shardingsphere.infra.database.type.dialect.PostgreSQLDatabaseType;
+import org.apache.shardingsphere.sharding.algorithm.keygen.SnowflakeKeyGenerateAlgorithm;
+import org.apache.shardingsphere.test.e2e.data.pipeline.cases.base.PipelineBaseE2EIT;
import org.apache.shardingsphere.test.e2e.data.pipeline.cases.migration.AbstractMigrationE2EIT;
import org.apache.shardingsphere.test.e2e.data.pipeline.cases.task.PostgreSQLIncrementTask;
import org.apache.shardingsphere.test.e2e.data.pipeline.env.enums.PipelineEnvTypeEnum;
import org.apache.shardingsphere.test.e2e.data.pipeline.framework.helper.PipelineCaseHelper;
import org.apache.shardingsphere.test.e2e.data.pipeline.framework.param.PipelineTestParameter;
-import org.apache.shardingsphere.sharding.algorithm.keygen.SnowflakeKeyGenerateAlgorithm;
-import org.apache.shardingsphere.test.e2e.data.pipeline.cases.base.PipelineBaseE2EIT;
+import org.apache.shardingsphere.test.e2e.data.pipeline.util.DataSourceExecuteUtil;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
-import org.springframework.jdbc.core.JdbcTemplate;
import java.sql.SQLException;
import java.time.LocalDateTime;
@@ -87,12 +87,11 @@ public final class PostgreSQLMigrationGeneralE2EIT extends AbstractMigrationE2EI
createTargetOrderTableRule();
createTargetOrderItemTableRule();
Pair<List<Object[]>, List<Object[]>> dataPair = PipelineCaseHelper.generateFullInsertData(testParam.getDatabaseType(), PipelineBaseE2EIT.TABLE_INIT_ROW_COUNT);
- JdbcTemplate jdbcTemplate = new JdbcTemplate(getSourceDataSource());
log.info("init data begin: {}", LocalDateTime.now());
- jdbcTemplate.batchUpdate(getExtraSQLCommand().getFullInsertOrder(getSourceTableOrderName()), dataPair.getLeft());
- jdbcTemplate.batchUpdate(getExtraSQLCommand().getFullInsertOrderItem(), dataPair.getRight());
+ DataSourceExecuteUtil.execute(getSourceDataSource(), getExtraSQLCommand().getFullInsertOrder(getSourceTableOrderName()), dataPair.getLeft());
+ DataSourceExecuteUtil.execute(getSourceDataSource(), getExtraSQLCommand().getFullInsertOrderItem(), dataPair.getRight());
log.info("init data end: {}", LocalDateTime.now());
- checkOrderMigration(jdbcTemplate);
+ checkOrderMigration();
checkOrderItemMigration();
for (String each : listJobId()) {
commitMigrationByJobId(each);
@@ -102,9 +101,9 @@ public final class PostgreSQLMigrationGeneralE2EIT extends AbstractMigrationE2EI
assertGreaterThanOrderTableInitRows(PipelineBaseE2EIT.TABLE_INIT_ROW_COUNT, PipelineBaseE2EIT.SCHEMA_NAME);
}
- private void checkOrderMigration(final JdbcTemplate jdbcTemplate) throws SQLException, InterruptedException {
+ private void checkOrderMigration() throws SQLException, InterruptedException {
startMigrationWithSchema(getSourceTableOrderName(), "t_order");
- startIncrementTask(new PostgreSQLIncrementTask(jdbcTemplate, PipelineBaseE2EIT.SCHEMA_NAME, getSourceTableOrderName(), 20));
+ startIncrementTask(new PostgreSQLIncrementTask(getSourceDataSource(), PipelineBaseE2EIT.SCHEMA_NAME, getSourceTableOrderName(), 20));
String jobId = getJobIdByTableName(getSourceTableOrderName());
waitIncrementTaskFinished(String.format("SHOW MIGRATION STATUS '%s'", jobId));
stopMigrationByJobId(jobId);
diff --git a/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/task/MySQLIncrementTask.java b/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/task/MySQLIncrementTask.java
index 28cf8be0d21..b0297b6dab1 100644
--- a/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/task/MySQLIncrementTask.java
+++ b/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/task/MySQLIncrementTask.java
@@ -22,8 +22,9 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.sharding.spi.KeyGenerateAlgorithm;
import org.apache.shardingsphere.test.e2e.data.pipeline.cases.base.BaseIncrementTask;
import org.apache.shardingsphere.test.e2e.data.pipeline.framework.helper.PipelineCaseHelper;
-import org.springframework.jdbc.core.JdbcTemplate;
+import org.apache.shardingsphere.test.e2e.data.pipeline.util.DataSourceExecuteUtil;
+import javax.sql.DataSource;
import java.time.Instant;
import java.util.concurrent.ThreadLocalRandom;
@@ -31,7 +32,7 @@ import java.util.concurrent.ThreadLocalRandom;
@Slf4j
public final class MySQLIncrementTask extends BaseIncrementTask {
- private final JdbcTemplate jdbcTemplate;
+ private final DataSource dataSource;
private final String orderTableName;
@@ -45,13 +46,14 @@ public final class MySQLIncrementTask extends BaseIncrementTask {
while (executeCount < executeCountLimit && !Thread.currentThread().isInterrupted()) {
Object orderPrimaryKey = insertOrder();
if (0 == executeCount % 2) {
- jdbcTemplate.update(String.format("DELETE FROM %s WHERE order_id = ?", orderTableName), orderPrimaryKey);
+ String sql = String.format("DELETE FROM %s WHERE order_id = ?", orderTableName);
+ DataSourceExecuteUtil.execute(dataSource, sql, new Object[]{orderPrimaryKey});
} else {
setNullToOrderFields(orderPrimaryKey);
updateOrderByPrimaryKey(orderPrimaryKey);
}
Object orderItemPrimaryKey = insertOrderItem();
- jdbcTemplate.update("UPDATE t_order_item SET status = ? WHERE item_id = ?", "updated" + Instant.now().getEpochSecond(), orderItemPrimaryKey);
+ DataSourceExecuteUtil.execute(dataSource, "UPDATE t_order_item SET status = ? WHERE item_id = ?", new Object[]{"updated" + Instant.now().getEpochSecond(), orderItemPrimaryKey});
executeCount++;
}
log.info("MySQL increment task runnable execute successfully.");
@@ -61,7 +63,7 @@ public final class MySQLIncrementTask extends BaseIncrementTask {
ThreadLocalRandom random = ThreadLocalRandom.current();
Object[] orderInsertDate = new Object[]{primaryKeyGenerateAlgorithm.generateKey(), random.nextInt(0, 6),
random.nextInt(1, 99), "中文测试"};
- jdbcTemplate.update(String.format("INSERT INTO %s (order_id,user_id,t_unsigned_int,status) VALUES (?, ?, ?, ?)", orderTableName), orderInsertDate);
+ DataSourceExecuteUtil.execute(dataSource, String.format("INSERT INTO %s (order_id,user_id,t_unsigned_int,status) VALUES (?, ?, ?, ?)", orderTableName), orderInsertDate);
return orderInsertDate[0];
}
@@ -69,18 +71,22 @@ public final class MySQLIncrementTask extends BaseIncrementTask {
ThreadLocalRandom random = ThreadLocalRandom.current();
String status = 0 == random.nextInt() % 2 ? null : "NOT-NULL";
Object[] orderInsertItemDate = new Object[]{primaryKeyGenerateAlgorithm.generateKey(), PipelineCaseHelper.generateSnowflakeKey(), random.nextInt(0, 6), status};
- jdbcTemplate.update("INSERT INTO t_order_item(item_id,order_id,user_id,status) VALUES(?, ?, ?, ?)", orderInsertItemDate);
+ DataSourceExecuteUtil.execute(dataSource, "INSERT INTO t_order_item(item_id,order_id,user_id,status) VALUES(?, ?, ?, ?)", orderInsertItemDate);
return orderInsertItemDate[0];
}
private void updateOrderByPrimaryKey(final Object primaryKey) {
Object[] updateData = {"updated" + Instant.now().getEpochSecond(), ThreadLocalRandom.current().nextInt(0, 100), primaryKey};
- jdbcTemplate.update(String.format("UPDATE %s SET t_char = ?,t_unsigned_int = ? WHERE order_id = ?", orderTableName), updateData);
+ DataSourceExecuteUtil.execute(dataSource, String.format("UPDATE %s SET t_char = ?,t_unsigned_int = ? WHERE order_id = ?", orderTableName), updateData);
// TODO 0000-00-00 00:00:00 now will cause consistency check failed.
// jdbcTemplate.update(String.format("UPDATE %s SET t_char = null,t_unsigned_int = 299,t_datetime='0000-00-00 00:00:00' WHERE order_id = ?", orderTableName), primaryKey);
+ /*
+ DataSourceUtil.execute(dataSource, String.format("UPDATE %s SET t_char = null,t_unsigned_int = 299,t_datetime='0000-00-00 00:00:00' WHERE order_id = ?", orderTableName),
+ new Object[]{primaryKey});
+ */
}
private void setNullToOrderFields(final Object primaryKey) {
- jdbcTemplate.update(String.format("UPDATE %s SET t_char = null, t_unsigned_int = null WHERE order_id = ?", orderTableName), primaryKey);
+ DataSourceExecuteUtil.execute(dataSource, String.format("UPDATE %s SET t_char = null, t_unsigned_int = null WHERE order_id = ?", orderTableName), new Object[]{primaryKey});
}
}
diff --git a/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/task/PostgreSQLIncrementTask.java b/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/task/PostgreSQLIncrementTask.java
index b18ba3a5583..69c0fd1bf29 100644
--- a/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/task/PostgreSQLIncrementTask.java
+++ b/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/cases/task/PostgreSQLIncrementTask.java
@@ -20,12 +20,13 @@ package org.apache.shardingsphere.test.e2e.data.pipeline.cases.task;
import com.google.common.base.Strings;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
-import org.apache.shardingsphere.test.e2e.data.pipeline.cases.base.BaseIncrementTask;
-import org.apache.shardingsphere.test.e2e.data.pipeline.framework.helper.PipelineCaseHelper;
import org.apache.shardingsphere.sharding.algorithm.keygen.SnowflakeKeyGenerateAlgorithm;
import org.apache.shardingsphere.sharding.spi.KeyGenerateAlgorithm;
-import org.springframework.jdbc.core.JdbcTemplate;
+import org.apache.shardingsphere.test.e2e.data.pipeline.cases.base.BaseIncrementTask;
+import org.apache.shardingsphere.test.e2e.data.pipeline.framework.helper.PipelineCaseHelper;
+import org.apache.shardingsphere.test.e2e.data.pipeline.util.DataSourceExecuteUtil;
+import javax.sql.DataSource;
import java.time.Instant;
import java.util.Properties;
import java.util.concurrent.ThreadLocalRandom;
@@ -36,7 +37,7 @@ public final class PostgreSQLIncrementTask extends BaseIncrementTask {
private static final KeyGenerateAlgorithm KEY_GENERATE_ALGORITHM = new SnowflakeKeyGenerateAlgorithm();
- private final JdbcTemplate jdbcTemplate;
+ private final DataSource dataSource;
private final String schema;
@@ -56,13 +57,15 @@ public final class PostgreSQLIncrementTask extends BaseIncrementTask {
while (executeCount < executeCountLimit && !Thread.currentThread().isInterrupted()) {
Object orderId = insertOrder();
if (0 == executeCount % 2) {
- jdbcTemplate.update(String.format("DELETE FROM %s WHERE order_id = ?", getTableNameWithSchema(orderTableName)), orderId);
+ DataSourceExecuteUtil.execute(dataSource, String.format("DELETE FROM %s WHERE order_id = ?", getTableNameWithSchema(orderTableName)), new Object[]{orderId});
} else {
updateOrderByPrimaryKey(orderId);
}
Object orderItemPrimaryKey = insertOrderItem();
- jdbcTemplate.update(String.format("UPDATE %s SET status = ? WHERE item_id = ?", getTableNameWithSchema("t_order_item")), "updated" + Instant.now().getEpochSecond(), orderItemPrimaryKey);
- jdbcTemplate.update(String.format("DELETE FROM %s WHERE item_id = ?", getTableNameWithSchema("t_order_item")), orderItemPrimaryKey);
+ String updateSql = String.format("UPDATE %s SET status = ? WHERE item_id = ?", getTableNameWithSchema("t_order_item"));
+ DataSourceExecuteUtil.execute(dataSource, updateSql, new Object[]{"updated" + Instant.now().getEpochSecond(), orderItemPrimaryKey});
+ String deleteSql = String.format("DELETE FROM %s WHERE item_id = ?", getTableNameWithSchema("t_order_item"));
+ DataSourceExecuteUtil.execute(dataSource, deleteSql, new Object[]{orderItemPrimaryKey});
executeCount++;
}
log.info("PostgreSQL increment task runnable execute successfully.");
@@ -74,7 +77,7 @@ public final class PostgreSQLIncrementTask extends BaseIncrementTask {
Object[] orderInsertDate = new Object[]{KEY_GENERATE_ALGORITHM.generateKey(), random.nextInt(0, 6), status};
String insertSQL = String.format("INSERT INTO %s (order_id,user_id,status) VALUES (?, ?, ?)", getTableNameWithSchema(orderTableName));
log.info("insert order sql:{}", insertSQL);
- jdbcTemplate.update(insertSQL, orderInsertDate);
+ DataSourceExecuteUtil.execute(dataSource, insertSQL, orderInsertDate);
return orderInsertDate[0];
}
@@ -82,13 +85,15 @@ public final class PostgreSQLIncrementTask extends BaseIncrementTask {
ThreadLocalRandom random = ThreadLocalRandom.current();
String status = 0 == random.nextInt() % 2 ? null : "NOT-NULL";
Object[] orderInsertItemDate = new Object[]{KEY_GENERATE_ALGORITHM.generateKey(), PipelineCaseHelper.generateSnowflakeKey(), random.nextInt(0, 6), status};
- jdbcTemplate.update(String.format("INSERT INTO %s(item_id,order_id,user_id,status) VALUES(?,?,?,?)", getTableNameWithSchema("t_order_item")), orderInsertItemDate);
+ String insertSql = "INSERT INTO %s(item_id,order_id,user_id,status) VALUES(?,?,?,?)";
+ DataSourceExecuteUtil.execute(dataSource, String.format(insertSql, getTableNameWithSchema("t_order_item")), orderInsertItemDate);
return orderInsertItemDate[0];
}
private void updateOrderByPrimaryKey(final Object primaryKey) {
Object[] updateData = {"updated" + Instant.now().getEpochSecond(), primaryKey};
- jdbcTemplate.update(String.format("UPDATE %s SET status = ? WHERE order_id = ?", getTableNameWithSchema(orderTableName)), updateData);
+ String updateSql = String.format("UPDATE %s SET status = ? WHERE order_id = ?", getTableNameWithSchema(orderTableName));
+ DataSourceExecuteUtil.execute(dataSource, String.format(updateSql, getTableNameWithSchema("t_order_item")), updateData);
}
private String getTableNameWithSchema(final String tableName) {
diff --git a/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/util/DataSourceExecuteUtil.java b/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/util/DataSourceExecuteUtil.java
new file mode 100644
index 00000000000..dcf803a20b6
--- /dev/null
+++ b/test/e2e/pipeline/src/test/java/org/apache/shardingsphere/test/e2e/data/pipeline/util/DataSourceExecuteUtil.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.shardingsphere.test.e2e.data.pipeline.util;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.List;
+
+/**
+ * Data source execute util.
+ */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public final class DataSourceExecuteUtil {
+
+ /**
+ * Execute sql.
+ *
+ * @param dataSource data source
+ * @param sql sql
+ */
+ public static void execute(final DataSource dataSource, final String sql) {
+ try (Connection connection = dataSource.getConnection()) {
+ connection.createStatement().execute(sql);
+ } catch (final SQLException ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
+ /**
+ * Execute sql.
+ *
+ * @param dataSource data source
+ * @param sql sql
+ * @param parameters parameters
+ */
+ public static void execute(final DataSource dataSource, final String sql, final Object[] parameters) {
+ try (Connection connection = dataSource.getConnection()) {
+ PreparedStatement preparedStatement = connection.prepareStatement(sql);
+ for (int i = 0; i < parameters.length; i++) {
+ preparedStatement.setObject(i + 1, parameters[i]);
+ }
+ preparedStatement.execute();
+ } catch (final SQLException ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+
+ /**
+ * Execute sql, batch mode.
+ *
+ * @param dataSource data source
+ * @param sql sql
+ * @param parameters parameters
+ */
+ public static void execute(final DataSource dataSource, final String sql, final List<Object[]> parameters) {
+ try (Connection connection = dataSource.getConnection()) {
+ PreparedStatement preparedStatement = connection.prepareStatement(sql);
+ for (Object[] each : parameters) {
+ for (int i = 0; i < each.length; i++) {
+ preparedStatement.setObject(i + 1, each[i]);
+ }
+ preparedStatement.addBatch();
+ }
+ preparedStatement.executeBatch();
+ } catch (final SQLException ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+}