You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by az...@apache.org on 2023/12/12 08:53:21 UTC
(shardingsphere) branch master updated: Fix sonar issue of pipeline (#29365)
This is an automated email from the ASF dual-hosted git repository.
azexin 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 ada5531fe08 Fix sonar issue of pipeline (#29365)
ada5531fe08 is described below
commit ada5531fe083c38f6d5b499037cc56b1a90d6b67
Author: Hongsheng Zhong <zh...@apache.org>
AuthorDate: Tue Dec 12 16:53:14 2023 +0800
Fix sonar issue of pipeline (#29365)
* Fix sonar issue of MemoryPipelineChannelCreator
* Fix sonar issue of AbstractSeparablePipelineJob
---
.../core/ingest/channel/memory/MemoryPipelineChannelCreator.java | 5 +++--
.../core/ingest/channel/memory/MultiplexMemoryPipelineChannel.java | 3 ++-
2 files changed, 5 insertions(+), 3 deletions(-)
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/memory/MemoryPipelineChannelCreator.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/memory/MemoryPipelineChannelCreator.java
index 91d2e78231b..46259a8c253 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/memory/MemoryPipelineChannelCreator.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/memory/MemoryPipelineChannelCreator.java
@@ -41,8 +41,9 @@ public final class MemoryPipelineChannelCreator implements PipelineChannelCreato
@Override
public PipelineChannel createPipelineChannel(final int outputConcurrency, final int averageElementSize, final AckCallback ackCallback) {
- return 1 == outputConcurrency ? new SimpleMemoryPipelineChannel(blockQueueSize / averageElementSize, ackCallback)
- : new MultiplexMemoryPipelineChannel(outputConcurrency, blockQueueSize < 1 ? 5 : blockQueueSize, ackCallback);
+ return 1 == outputConcurrency
+ ? new SimpleMemoryPipelineChannel(blockQueueSize / averageElementSize, ackCallback)
+ : new MultiplexMemoryPipelineChannel(outputConcurrency, blockQueueSize, ackCallback);
}
@Override
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/memory/MultiplexMemoryPipelineChannel.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/memory/MultiplexMemoryPipelineChannel.java
index a829030d5ff..da247ee2b18 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/memory/MultiplexMemoryPipelineChannel.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/ingest/channel/memory/MultiplexMemoryPipelineChannel.java
@@ -46,7 +46,8 @@ public final class MultiplexMemoryPipelineChannel implements PipelineChannel {
public MultiplexMemoryPipelineChannel(final int channelNumber, final int blockQueueSize, final AckCallback ackCallback) {
this.channelNumber = channelNumber;
- channels = IntStream.range(0, channelNumber).mapToObj(each -> new SimpleMemoryPipelineChannel(blockQueueSize, ackCallback)).collect(Collectors.toList());
+ int handledQueueSize = blockQueueSize < 1 ? 5 : blockQueueSize;
+ channels = IntStream.range(0, channelNumber).mapToObj(each -> new SimpleMemoryPipelineChannel(handledQueueSize, ackCallback)).collect(Collectors.toList());
}
@Override