You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by la...@apache.org on 2023/11/30 07:56:09 UTC
(iotdb) branch fixMemory created (now f8428d3bdfd)
This is an automated email from the ASF dual-hosted git repository.
lancelly pushed a change to branch fixMemory
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at f8428d3bdfd add close
This branch includes the following new commits:
new 7e3d1155f8c replace Validate.notNull and add Logger.isDebugEnabled
new f8428d3bdfd add close
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 02/02: add close
Posted by la...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lancelly pushed a commit to branch fixMemory
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f8428d3bdfdbc988d9a137913c8ff421aef473ed
Author: lancelly <14...@qq.com>
AuthorDate: Thu Nov 30 15:47:21 2023 +0800
add close
---
.../queryengine/execution/operator/process/TopKOperator.java | 10 ++++++++++
1 file changed, 10 insertions(+)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TopKOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TopKOperator.java
index d14741a31d6..d669d1ab010 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TopKOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TopKOperator.java
@@ -209,6 +209,16 @@ public class TopKOperator implements ProcessOperator {
return null;
}
+ @Override
+ public void close() throws Exception {
+ for (Operator operator : deviceOperators) {
+ if (operator != null) {
+ operator.close();
+ }
+ }
+ deviceOperators.clear();
+ }
+
@Override
public long calculateMaxPeekMemory() {
// traverse each child serial,
(iotdb) 01/02: replace Validate.notNull and add Logger.isDebugEnabled
Posted by la...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lancelly pushed a commit to branch fixMemory
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 7e3d1155f8c0e847a9e1a08d29a0eb8afd70c9fc
Author: lancelly <14...@qq.com>
AuthorDate: Wed Nov 29 22:03:59 2023 +0800
replace Validate.notNull and add Logger.isDebugEnabled
---
.../db/queryengine/execution/driver/Driver.java | 4 --
.../execution/exchange/sink/LocalSinkChannel.java | 39 ++++++++----
.../execution/exchange/sink/ShuffleSinkHandle.java | 27 +++++---
.../execution/exchange/sink/SinkChannel.java | 73 ++++++++++++++--------
.../exchange/source/LocalSourceHandle.java | 15 +++--
.../execution/exchange/source/SourceHandle.java | 20 +++---
.../fragment/FragmentInstanceExecution.java | 3 +-
.../queryengine/execution/memory/MemoryPool.java | 6 +-
.../schedule/DriverTaskTimeoutSentinelThread.java | 2 +-
9 files changed, 120 insertions(+), 69 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/Driver.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/Driver.java
index 45b897d9c60..61b3a634e82 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/Driver.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/Driver.java
@@ -115,10 +115,6 @@ public abstract class Driver implements IDriver {
/** release resource this driver used. */
protected abstract void releaseResource();
- public int getDependencyDriverIndex() {
- return driverContext.getDependencyDriverIndex();
- }
-
@Override
public ListenableFuture<?> processFor(Duration duration) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/LocalSinkChannel.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/LocalSinkChannel.java
index d23f7c04450..2fe94bbeb51 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/LocalSinkChannel.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/LocalSinkChannel.java
@@ -57,8 +57,8 @@ public class LocalSinkChannel implements ISinkChannel {
DataExchangeCostMetricSet.getInstance();
public LocalSinkChannel(SharedTsBlockQueue queue, SinkListener sinkListener) {
- this.sinkListener = Validate.notNull(sinkListener);
- this.queue = Validate.notNull(queue);
+ this.sinkListener = Validate.notNull(sinkListener, "sinkListener can not be null.");
+ this.queue = Validate.notNull(queue, "queue can not be null.");
this.queue.setSinkChannel(this);
blocked = queue.getCanAddTsBlock();
}
@@ -67,9 +67,10 @@ public class LocalSinkChannel implements ISinkChannel {
TFragmentInstanceId localFragmentInstanceId,
SharedTsBlockQueue queue,
SinkListener sinkListener) {
- this.localFragmentInstanceId = Validate.notNull(localFragmentInstanceId);
- this.sinkListener = Validate.notNull(sinkListener);
- this.queue = Validate.notNull(queue);
+ this.localFragmentInstanceId =
+ Validate.notNull(localFragmentInstanceId, "localFragmentInstanceId can not be null.");
+ this.sinkListener = Validate.notNull(sinkListener, "sinkListener can not be null.");
+ this.queue = Validate.notNull(queue, "queue can not be null.");
this.queue.setSinkChannel(this);
// SinkChannel can send data after SourceHandle asks it to
blocked = queue.getCanAddTsBlock();
@@ -140,7 +141,9 @@ public class LocalSinkChannel implements ISinkChannel {
if (queue.hasNoMoreTsBlocks()) {
return;
}
- LOGGER.debug("[StartSendTsBlockOnLocal]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[StartSendTsBlockOnLocal]");
+ }
synchronized (this) {
blocked = queue.add(tsBlock);
}
@@ -155,7 +158,9 @@ public class LocalSinkChannel implements ISinkChannel {
public void setNoMoreTsBlocks() {
synchronized (queue) {
synchronized (this) {
- LOGGER.debug("[StartSetNoMoreTsBlocksOnLocal]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[StartSetNoMoreTsBlocksOnLocal]");
+ }
if (aborted || closed) {
return;
}
@@ -164,12 +169,16 @@ public class LocalSinkChannel implements ISinkChannel {
}
}
checkAndInvokeOnFinished();
- LOGGER.debug("[EndSetNoMoreTsBlocksOnLocal]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[EndSetNoMoreTsBlocksOnLocal]");
+ }
}
@Override
public void abort() {
- LOGGER.debug("[StartAbortLocalSinkChannel]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[StartAbortLocalSinkChannel]");
+ }
synchronized (queue) {
synchronized (this) {
if (aborted || closed) {
@@ -184,12 +193,16 @@ public class LocalSinkChannel implements ISinkChannel {
}
}
}
- LOGGER.debug("[EndAbortLocalSinkChannel]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[EndAbortLocalSinkChannel]");
+ }
}
@Override
public void close() {
- LOGGER.debug("[StartCloseLocalSinkChannel]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[StartCloseLocalSinkChannel]");
+ }
synchronized (queue) {
synchronized (this) {
if (aborted || closed) {
@@ -203,7 +216,9 @@ public class LocalSinkChannel implements ISinkChannel {
}
}
}
- LOGGER.debug("[EndCloseLocalSinkChannel]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[EndCloseLocalSinkChannel]");
+ }
}
public SharedTsBlockQueue getSharedTsBlockQueue() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/ShuffleSinkHandle.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/ShuffleSinkHandle.java
index 234fe2dbecb..7c31ab67e04 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/ShuffleSinkHandle.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/ShuffleSinkHandle.java
@@ -75,10 +75,13 @@ public class ShuffleSinkHandle implements ISinkHandle {
DownStreamChannelIndex downStreamChannelIndex,
ShuffleStrategyEnum shuffleStrategyEnum,
MPPDataExchangeManager.SinkListener sinkListener) {
- this.localFragmentInstanceId = Validate.notNull(localFragmentInstanceId);
- this.downStreamChannelList = Validate.notNull(downStreamChannelList);
- this.downStreamChannelIndex = Validate.notNull(downStreamChannelIndex);
- this.sinkListener = Validate.notNull(sinkListener);
+ this.localFragmentInstanceId =
+ Validate.notNull(localFragmentInstanceId, "localFragmentInstanceId can not be null.");
+ this.downStreamChannelList =
+ Validate.notNull(downStreamChannelList, "downStreamChannelList can not be null.");
+ this.downStreamChannelIndex =
+ Validate.notNull(downStreamChannelIndex, "downStreamChannelIndex can not be null.");
+ this.sinkListener = Validate.notNull(sinkListener, "sinkListener can not be null.");
this.channelNum = downStreamChannelList.size();
this.shuffleStrategy = getShuffleStrategy(shuffleStrategyEnum);
this.hasSetNoMoreTsBlocks = new boolean[channelNum];
@@ -186,7 +189,9 @@ public class ShuffleSinkHandle implements ISinkHandle {
return;
}
aborted = true;
- LOGGER.debug("[StartAbortShuffleSinkHandle]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[StartAbortShuffleSinkHandle]");
+ }
boolean meetError = false;
Exception firstException = null;
for (ISink channel : downStreamChannelList) {
@@ -203,7 +208,9 @@ public class ShuffleSinkHandle implements ISinkHandle {
LOGGER.warn("Error occurred when try to abort channel.", firstException);
}
sinkListener.onAborted(this);
- LOGGER.debug("[EndAbortShuffleSinkHandle]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[EndAbortShuffleSinkHandle]");
+ }
}
// Add synchronized on this method may lead to Dead Lock
@@ -216,7 +223,9 @@ public class ShuffleSinkHandle implements ISinkHandle {
return;
}
closed = true;
- LOGGER.debug("[StartCloseShuffleSinkHandle]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[StartCloseShuffleSinkHandle]");
+ }
boolean meetError = false;
Exception firstException = null;
for (ISink channel : downStreamChannelList) {
@@ -233,7 +242,9 @@ public class ShuffleSinkHandle implements ISinkHandle {
LOGGER.warn("Error occurred when try to close channel.", firstException);
}
sinkListener.onFinish(this);
- LOGGER.debug("[EndCloseShuffleSinkHandle]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[EndCloseShuffleSinkHandle]");
+ }
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/SinkChannel.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/SinkChannel.java
index 0230ee9d0aa..4290f74ba85 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/SinkChannel.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/sink/SinkChannel.java
@@ -135,17 +135,20 @@ public class SinkChannel implements ISinkChannel {
SinkListener sinkListener,
IClientManager<TEndPoint, SyncDataNodeMPPDataExchangeServiceClient>
mppDataExchangeServiceClientManager) {
- this.remoteEndpoint = Validate.notNull(remoteEndpoint);
- this.remoteFragmentInstanceId = Validate.notNull(remoteFragmentInstanceId);
- this.remotePlanNodeId = Validate.notNull(remotePlanNodeId);
- this.localPlanNodeId = Validate.notNull(localPlanNodeId);
- this.localFragmentInstanceId = Validate.notNull(localFragmentInstanceId);
+ this.remoteEndpoint = Validate.notNull(remoteEndpoint, "remoteEndPoint can not be null.");
+ this.remoteFragmentInstanceId =
+ Validate.notNull(remoteFragmentInstanceId, "remoteFragmentInstanceId can not be null.");
+ this.remotePlanNodeId = Validate.notNull(remotePlanNodeId, "remotePlanNodeId can not be null.");
+ this.localPlanNodeId = Validate.notNull(localPlanNodeId, "localPlanNodeId can not be null.");
+ this.localFragmentInstanceId =
+ Validate.notNull(localFragmentInstanceId, "localFragmentInstanceId can not be null.");
this.fullFragmentInstanceId =
FragmentInstanceId.createFragmentInstanceIdFromTFragmentInstanceId(localFragmentInstanceId);
- this.localMemoryManager = Validate.notNull(localMemoryManager);
- this.executorService = Validate.notNull(executorService);
- this.serde = Validate.notNull(serde);
- this.sinkListener = Validate.notNull(sinkListener);
+ this.localMemoryManager =
+ Validate.notNull(localMemoryManager, "localMemoryManager can not be null.");
+ this.executorService = Validate.notNull(executorService, "executorService can not be null.");
+ this.serde = Validate.notNull(serde, "serde can not be null.");
+ this.sinkListener = Validate.notNull(sinkListener, "sinkListener can not be null.");
this.mppDataExchangeServiceClientManager = mppDataExchangeServiceClientManager;
this.retryIntervalInMs = DEFAULT_RETRY_INTERVAL_IN_MS;
this.threadName =
@@ -221,7 +224,9 @@ public class SinkChannel implements ISinkChannel {
@Override
public synchronized void setNoMoreTsBlocks() {
- LOGGER.debug("[StartSetNoMoreTsBlocks]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[StartSetNoMoreTsBlocks]");
+ }
if (aborted || closed) {
return;
}
@@ -230,7 +235,9 @@ public class SinkChannel implements ISinkChannel {
@Override
public synchronized void abort() {
- LOGGER.debug("[StartAbortSinkChannel]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[StartAbortSinkChannel]");
+ }
if (aborted || closed) {
return;
}
@@ -250,12 +257,16 @@ public class SinkChannel implements ISinkChannel {
}
sinkListener.onAborted(this);
aborted = true;
- LOGGER.debug("[EndAbortSinkChannel]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[EndAbortSinkChannel]");
+ }
}
@Override
public synchronized void close() {
- LOGGER.debug("[StartCloseSinkChannel]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[StartCloseSinkChannel]");
+ }
if (closed || aborted) {
return;
}
@@ -275,7 +286,9 @@ public class SinkChannel implements ISinkChannel {
}
invokeOnFinished();
closed = true;
- LOGGER.debug("[EndCloseSinkChannel]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[EndCloseSinkChannel]");
+ }
}
private void invokeOnFinished() {
@@ -310,10 +323,12 @@ public class SinkChannel implements ISinkChannel {
public synchronized ByteBuffer getSerializedTsBlock(int sequenceId) throws IOException {
if (aborted || closed) {
- LOGGER.debug(
- "SinkChannel still receive getting TsBlock request after being aborted={} or closed={}",
- aborted,
- closed);
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug(
+ "SinkChannel still receive getting TsBlock request after being aborted={} or closed={}",
+ aborted,
+ closed);
+ }
throw new GetTsBlockFromClosedOrAbortedChannelException("SinkChannel is aborted or closed. ");
}
Pair<TsBlock, Long> pair = sequenceIdToTsBlock.get(sequenceId);
@@ -347,7 +362,9 @@ public class SinkChannel implements ISinkChannel {
freedBytes += entry.getValue().right;
bufferRetainedSizeInBytes -= entry.getValue().right;
iterator.remove();
- LOGGER.debug("[ACKTsBlock] {}.", entry.getKey());
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[ACKTsBlock] {}.", entry.getKey());
+ }
}
// there may exist duplicate ack message in network caused by caller retrying, if so duplicate
@@ -457,12 +474,14 @@ public class SinkChannel implements ISinkChannel {
@Override
public void run() {
try (SetThreadName sinkChannelName = new SetThreadName(threadName)) {
- LOGGER.debug(
- "[NotifyNewTsBlock] [{}, {}) to {}.{}",
- startSequenceId,
- startSequenceId + blockSizes.size(),
- remoteFragmentInstanceId,
- remotePlanNodeId);
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug(
+ "[NotifyNewTsBlock] [{}, {}) to {}.{}",
+ startSequenceId,
+ startSequenceId + blockSizes.size(),
+ remoteFragmentInstanceId,
+ remotePlanNodeId);
+ }
int attempt = 0;
TNewDataBlockEvent newDataBlockEvent =
new TNewDataBlockEvent(
@@ -509,7 +528,9 @@ public class SinkChannel implements ISinkChannel {
@Override
public void run() {
try (SetThreadName sinkChannelName = new SetThreadName(threadName)) {
- LOGGER.debug("[NotifyNoMoreTsBlock]");
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("[NotifyNoMoreTsBlock]");
+ }
int attempt = 0;
TEndOfDataBlockEvent endOfDataBlockEvent =
new TEndOfDataBlockEvent(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/LocalSourceHandle.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/LocalSourceHandle.java
index 51ac04d5fea..de37c64e460 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/LocalSourceHandle.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/LocalSourceHandle.java
@@ -64,9 +64,10 @@ public class LocalSourceHandle implements ISourceHandle {
public LocalSourceHandle(
SharedTsBlockQueue queue, SourceHandleListener sourceHandleListener, String threadName) {
- this.queue = Validate.notNull(queue);
+ this.queue = Validate.notNull(queue, "queue can not be null.");
this.queue.setSourceHandle(this);
- this.sourceHandleListener = Validate.notNull(sourceHandleListener);
+ this.sourceHandleListener =
+ Validate.notNull(sourceHandleListener, "sourceHandleListener can not be null.");
this.threadName = threadName;
}
@@ -76,11 +77,13 @@ public class LocalSourceHandle implements ISourceHandle {
String localPlanNodeId,
SharedTsBlockQueue queue,
SourceHandleListener sourceHandleListener) {
- this.localFragmentInstanceId = Validate.notNull(localFragmentInstanceId);
- this.localPlanNodeId = Validate.notNull(localPlanNodeId);
- this.queue = Validate.notNull(queue);
+ this.localFragmentInstanceId =
+ Validate.notNull(localFragmentInstanceId, "localFragmentInstanceId can not be null.");
+ this.localPlanNodeId = Validate.notNull(localPlanNodeId, "localPlanNodeId can not be null.");
+ this.queue = Validate.notNull(queue, "queue can not be null.");
this.queue.setSourceHandle(this);
- this.sourceHandleListener = Validate.notNull(sourceHandleListener);
+ this.sourceHandleListener =
+ Validate.notNull(sourceHandleListener, "sourceHandleListener can not be null.");
this.threadName = createFullIdFrom(localFragmentInstanceId, localPlanNodeId);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/SourceHandle.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/SourceHandle.java
index 500ac1ad417..e7c25d14c04 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/SourceHandle.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/exchange/source/SourceHandle.java
@@ -132,17 +132,21 @@ public class SourceHandle implements ISourceHandle {
SourceHandleListener sourceHandleListener,
IClientManager<TEndPoint, SyncDataNodeMPPDataExchangeServiceClient>
mppDataExchangeServiceClientManager) {
- this.remoteEndpoint = Validate.notNull(remoteEndpoint);
- this.remoteFragmentInstanceId = Validate.notNull(remoteFragmentInstanceId);
- this.localFragmentInstanceId = Validate.notNull(localFragmentInstanceId);
+ this.remoteEndpoint = Validate.notNull(remoteEndpoint, "remoteEndpoint can not be null.");
+ this.remoteFragmentInstanceId =
+ Validate.notNull(remoteFragmentInstanceId, "remoteFragmentInstanceId can not be null.");
+ this.localFragmentInstanceId =
+ Validate.notNull(localFragmentInstanceId, "localFragmentInstanceId can not be null.");
this.fullFragmentInstanceId =
FragmentInstanceId.createFragmentInstanceIdFromTFragmentInstanceId(localFragmentInstanceId);
- this.localPlanNodeId = Validate.notNull(localPlanNodeId);
+ this.localPlanNodeId = Validate.notNull(localPlanNodeId, "localPlanNodeId can not be null.");
this.indexOfUpstreamSinkHandle = indexOfUpstreamSinkHandle;
- this.localMemoryManager = Validate.notNull(localMemoryManager);
- this.executorService = Validate.notNull(executorService);
- this.serde = Validate.notNull(serde);
- this.sourceHandleListener = Validate.notNull(sourceHandleListener);
+ this.localMemoryManager =
+ Validate.notNull(localMemoryManager, "localMemoryManager can not be null.");
+ this.executorService = Validate.notNull(executorService, "executorService can not be null.");
+ this.serde = Validate.notNull(serde, "serde can not be null.");
+ this.sourceHandleListener =
+ Validate.notNull(sourceHandleListener, "sourceHandleListener can not be null.");
this.bufferRetainedSizeInBytes = 0L;
this.mppDataExchangeServiceClientManager = mppDataExchangeServiceClientManager;
this.retryIntervalInMs = DEFAULT_RETRY_INTERVAL_IN_MS;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecution.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecution.java
index 840617a512f..890d1098353 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecution.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecution.java
@@ -165,7 +165,8 @@ public class FragmentInstanceExecution {
} catch (Throwable t) {
try (SetThreadName threadName = new SetThreadName(instanceId.getFullId())) {
LOGGER.error(
- "Errors happened while trying to finish FI, resource may already leak!", t);
+ "Errors occurred while attempting to finish the FI process, potentially leading to resource leakage.",
+ t);
}
}
});
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/memory/MemoryPool.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/memory/MemoryPool.java
index 936a4cbcb27..3b8adc70d44 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/memory/MemoryPool.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/memory/MemoryPool.java
@@ -123,7 +123,7 @@ public class MemoryPool {
new ConcurrentLinkedQueue<>();
public MemoryPool(String id, long maxBytes, long maxBytesPerFragmentInstance) {
- this.id = Validate.notNull(id);
+ this.id = Validate.notNull(id, "id can not be null.");
Validate.isTrue(maxBytes > 0L, "max bytes should be greater than zero: %d", maxBytes);
this.maxBytes = maxBytes;
Validate.isTrue(
@@ -291,7 +291,7 @@ public class MemoryPool {
// add synchronized on the future to avoid that the future is concurrently completed by
// MemoryPool.free() which may lead to memory leak.
synchronized (future) {
- Validate.notNull(future);
+ Validate.notNull(future, "The future to be cancelled can not be null.");
// If the future is not a MemoryReservationFuture, it must have been completed.
if (future.isDone()) {
return 0L;
@@ -305,7 +305,7 @@ public class MemoryPool {
}
public void free(String queryId, String fragmentInstanceId, String planNodeId, long bytes) {
- Validate.notNull(queryId);
+ Validate.notNull(queryId, "queryId can not be null.");
Validate.isTrue(bytes > 0L);
try {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskTimeoutSentinelThread.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskTimeoutSentinelThread.java
index f7bbfab725d..9fb4e59dda2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskTimeoutSentinelThread.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/schedule/DriverTaskTimeoutSentinelThread.java
@@ -31,7 +31,7 @@ public class DriverTaskTimeoutSentinelThread extends AbstractDriverThread {
private static final Logger LOGGER =
LoggerFactory.getLogger(DriverTaskTimeoutSentinelThread.class);
- private final long SLEEP_BOUND = 5 * 1000L;
+ private static final long SLEEP_BOUND = 5 * 1000L;
public DriverTaskTimeoutSentinelThread(
String workerId,