You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by me...@apache.org on 2025/03/24 01:18:16 UTC
(shardingsphere) branch master updated: Review and improve pipeline code (#35067)
This is an automated email from the ASF dual-hosted git repository.
menghaoran 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 6eea8cd2cce Review and improve pipeline code (#35067)
6eea8cd2cce is described below
commit 6eea8cd2ccefc2332a42b8037805b29dec97a0e0
Author: Hongsheng Zhong <zh...@apache.org>
AuthorDate: Mon Mar 24 09:18:10 2025 +0800
Review and improve pipeline code (#35067)
---
.../pipeline/core/ingest/dumper/inventory/InventoryDumper.java | 7 ++++---
.../data/pipeline/core/job/service/PipelineJobManager.java | 2 +-
.../mysql/ingest/incremental/client/MySQLBinlogClient.java | 2 +-
.../mysql/ingest/incremental/client/MySQLBinlogClientTest.java | 2 +-
4 files changed, 7 insertions(+), 6 deletions(-)
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/InventoryDumper.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/InventoryDumper.java
index b744834efee..c4dc73a969f 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/InventoryDumper.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/dumper/inventory/InventoryDumper.java
@@ -203,15 +203,16 @@ public final class InventoryDumper extends AbstractPipelineLifecycleRunnable imp
private void setParameters(final PreparedStatement preparedStatement, final InventoryQueryParameter<?> queryParam) throws SQLException {
if (queryParam instanceof InventoryRangeQueryParameter) {
+ int parameterIndex = 1;
Object lower = ((InventoryRangeQueryParameter) queryParam).getValue().getLower();
if (null != lower) {
- preparedStatement.setObject(1, lower);
+ preparedStatement.setObject(parameterIndex++, lower);
}
Object upper = ((InventoryRangeQueryParameter) queryParam).getValue().getUpper();
if (null != upper) {
- preparedStatement.setObject(2, upper);
+ preparedStatement.setObject(parameterIndex++, upper);
}
- preparedStatement.setInt(3, dumperContext.getBatchSize());
+ preparedStatement.setInt(parameterIndex, dumperContext.getBatchSize());
} else if (queryParam instanceof InventoryPointQueryParameter) {
preparedStatement.setObject(1, queryParam.getValue());
} else {
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java
index c9d11ea3a7d..9420c3d0ac1 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/service/PipelineJobManager.java
@@ -186,7 +186,7 @@ public final class PipelineJobManager {
public Collection<ShardingInfo> getJobShardingInfos(final PipelineContextKey contextKey, final String jobId) {
try {
return PipelineAPIFactory.getShardingStatisticsAPI(contextKey).getShardingInfo(jobId);
- } catch (final NullPointerException | UnsupportedOperationException ex) {
+ } catch (final UnsupportedOperationException ex) {
return Collections.emptyList();
}
}
diff --git a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClient.java b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClient.java
index ac2b76ee971..ea6bbfd6f24 100644
--- a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClient.java
+++ b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClient.java
@@ -189,7 +189,7 @@ public final class MySQLBinlogClient {
}
private void configureHeartbeat() {
- execute("SET @master_heartbeat_period=15000000000");
+ execute("SET @master_heartbeat_period=" + TimeUnit.SECONDS.toNanos(15L));
}
private void registerSlave() {
diff --git a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClientTest.java b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClientTest.java
index be0a32076a0..7f6e0017e3b 100644
--- a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClientTest.java
+++ b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClientTest.java
@@ -144,7 +144,7 @@ class MySQLBinlogClientTest {
} catch (final ReflectiveOperationException ex) {
throw new RuntimeException(ex);
}
- if (null != responseCallback) {
+ if (null != responseCallback && !responseCallback.isDone()) {
responseCallback.setSuccess(response);
}
TimeUnit.SECONDS.sleep(1L);