This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch delete_with_attribute in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 63303aada1be6fcd173cb93ff7d0923054e34cfe Author: Tian Jiang <[email protected]> AuthorDate: Fri Jun 26 14:44:49 2026 +0800 temp1 --- .../iotdb/db/i18n/StorageEngineMessages.java | 2 + .../iotdb/db/i18n/StorageEngineMessages.java | 2 + .../DirectBufferMemoryAllocationException.java | 35 ++++++++ .../iotdb/db/storageengine/StorageEngine.java | 7 ++ .../db/storageengine/dataregion/DataRegion.java | 17 +--- .../iotdb/db/storageengine/StorageEngineTest.java | 97 ++++++++++++++++++++++ 6 files changed, 147 insertions(+), 13 deletions(-) diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java index cc708e7e48c..a7a5bc6c9c3 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java @@ -80,6 +80,8 @@ public final class StorageEngineMessages { public static final String INTERRUPTED_WAITING_THREAD_POOL_EXIT = "Interrupted while waiting {} thread pool to exit. "; public static final String BUFFERED_ARRAY_SIZE_THRESHOLD = "BufferedArraySizeThreshold is {}"; public static final String CURRENT_SG_COST = "Current Sg cost is {}"; + public static final String DIRECT_BUFFER_MEMORY_ALLOCATION_FAILED = + "Total allocated memory for direct buffer will be %d, which is greater than limit mem cost: %d"; public static final String FORCE_DEGRADE_TSFILE_RESOURCE = "Force degrade tsfile resource {}"; public static final String CANNOT_DEGRADE_TIME_INDEX_ALL_FILE_LEVEL = "Can't degrade time index any more because all time index are file level."; public static final String DEGRADE_TSFILE_RESOURCE = "Degrade tsfile resource {}"; diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java index 792909a97e0..4addaeeaa6c 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java @@ -80,6 +80,8 @@ public final class StorageEngineMessages { public static final String INTERRUPTED_WAITING_THREAD_POOL_EXIT = "等待 {} 线程池退出时被中断。 "; public static final String BUFFERED_ARRAY_SIZE_THRESHOLD = "BufferedArraySizeThreshold 为 {}"; public static final String CURRENT_SG_COST = "当前存储组内存开销为 {}"; + public static final String DIRECT_BUFFER_MEMORY_ALLOCATION_FAILED = + "DirectBuffer 总分配内存将达到 %d，超过内存限制：%d"; public static final String FORCE_DEGRADE_TSFILE_RESOURCE = "强制降级 TsFile 资源 {}"; public static final String CANNOT_DEGRADE_TIME_INDEX_ALL_FILE_LEVEL = "无法继续降级时间索引，所有时间索引已为文件级别。"; public static final String DEGRADE_TSFILE_RESOURCE = "降级 TsFile 资源 {}"; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/DirectBufferMemoryAllocationException.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/DirectBufferMemoryAllocationException.java new file mode 100644 index 00000000000..0e29e76b880 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/DirectBufferMemoryAllocationException.java @@ -0,0 +1,35 @@ +/* + * 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.iotdb.db.exception; + +import org.apache.iotdb.db.i18n.StorageEngineMessages; + +public class DirectBufferMemoryAllocationException extends DataRegionException { + + private static final long serialVersionUID = -8268138833145785933L; + + public DirectBufferMemoryAllocationException( + long allocatedDirectBufferMemory, long directBufferMemoryLimit) { + super( + String.format( + StorageEngineMessages.DIRECT_BUFFER_MEMORY_ALLOCATION_FAILED, + allocatedDirectBufferMemory, + directBufferMemoryLimit)); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java index 532b6878577..389297ac401 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java @@ -49,6 +49,7 @@ import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.DataRegionException; +import org.apache.iotdb.db.exception.DirectBufferMemoryAllocationException; import org.apache.iotdb.db.exception.StorageEngineException; import org.apache.iotdb.db.exception.TsFileProcessorException; import org.apache.iotdb.db.exception.WriteProcessRejectException; @@ -260,6 +261,12 @@ public class StorageEngine implements IService { } catch (DataRegionException e) { LOGGER.error( "Failed to recover data region {}[{}]", sgName, dataRegionId.getId(), e); + WALRecoverManager.getInstance() + .getAllDataRegionScannedLatch() + .countDownWithException(e.getMessage()); + if (e instanceof DirectBufferMemoryAllocationException) { + throw e; + } return null; } dataRegionMap.put(dataRegionId, dataRegion); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index 03807c60d74..f98ba5ad577 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -61,6 +61,7 @@ import org.apache.iotdb.db.consensus.DataRegionConsensusImpl; import org.apache.iotdb.db.exception.BatchProcessException; import org.apache.iotdb.db.exception.DataRegionException; import org.apache.iotdb.db.exception.DataTypeInconsistentException; +import org.apache.iotdb.db.exception.DirectBufferMemoryAllocationException; import org.apache.iotdb.db.exception.TsFileProcessorException; import org.apache.iotdb.db.exception.WriteProcessException; import org.apache.iotdb.db.exception.WriteProcessRejectException; @@ -594,10 +595,6 @@ public class DataRegion implements IDataRegionForQuery { try { recoverCompaction(); } catch (Exception e) { - // signal wal recover manager to recover this region's files - WALRecoverManager.getInstance() - .getAllDataRegionScannedLatch() - .countDownWithException(e.getMessage()); throw new DataRegionException(e); } @@ -768,10 +765,6 @@ public class DataRegion implements IDataRegionForQuery { updatePartitionFileVersion(partitionNum, resource.getVersion()); } } catch (IOException e) { - // signal wal recover manager to recover this region's files - WALRecoverManager.getInstance() - .getAllDataRegionScannedLatch() - .countDownWithException(e.getMessage()); throw new DataRegionException(e); } @@ -5117,11 +5110,9 @@ public class DataRegion implements IDataRegionForQuery { private void acquireDirectBufferMemory() throws DataRegionException { long acquireDirectBufferMemCost = getAcquireDirectBufferMemCost(); if (!SystemInfo.getInstance().addDirectBufferMemoryCost(acquireDirectBufferMemCost)) { - throw new DataRegionException( - "Total allocated memory for direct buffer will be " - + (SystemInfo.getInstance().getDirectBufferMemoryCost() + acquireDirectBufferMemCost) - + ", which is greater than limit mem cost: " - + SystemInfo.getInstance().getTotalDirectBufferMemorySizeLimit()); + throw new DirectBufferMemoryAllocationException( + SystemInfo.getInstance().getDirectBufferMemoryCost() + acquireDirectBufferMemCost, + SystemInfo.getInstance().getTotalDirectBufferMemorySizeLimit()); } this.directBufferMemoryCost = acquireDirectBufferMemCost; } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/StorageEngineTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/StorageEngineTest.java index 4c4ac20e9ae..eb38bdff25a 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/StorageEngineTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/StorageEngineTest.java @@ -18,9 +18,13 @@ */ package org.apache.iotdb.db.storageengine; +import org.apache.iotdb.commons.concurrent.ExceptionalCountDownLatch; import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.utils.TimePartitionUtils; +import org.apache.iotdb.db.exception.DataRegionException; +import org.apache.iotdb.db.exception.DirectBufferMemoryAllocationException; import org.apache.iotdb.db.storageengine.dataregion.DataRegion; +import org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRecoverManager; import com.google.common.collect.Lists; import org.junit.After; @@ -28,12 +32,21 @@ import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.ArgumentMatchers; import org.powermock.api.mockito.PowerMockito; import org.powermock.core.classloader.annotations.PowerMockIgnore; import org.powermock.core.classloader.annotations.PrepareForTest; import org.powermock.modules.junit4.PowerMockRunner; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Collections; import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; @PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"}) @RunWith(PowerMockRunner.class) @@ -84,4 +97,88 @@ public class StorageEngineTest { Assert.assertEquals(1, TimePartitionUtils.getTimePartitionId(timePartitionInterval * 2 - 1)); Assert.assertEquals(2, TimePartitionUtils.getTimePartitionId(timePartitionInterval * 2 + 1)); } + + @Test + public void testNotifyWALRecoverManagerWhenDirectBufferAllocationFailed() throws Exception { + StorageEngine spyStorageEngine = PowerMockito.spy(storageEngine); + DirectBufferMemoryAllocationException directBufferMemoryAllocationException = + new DirectBufferMemoryAllocationException(2, 1); + ExecutorService testThreadPool = Executors.newSingleThreadExecutor(); + setCachedThreadPool(spyStorageEngine, testThreadPool); + PowerMockito.doReturn( + Collections.singletonMap("root.sg", Lists.newArrayList(new DataRegionId(0)))) + .when(spyStorageEngine) + .getLocalDataRegionInfo(); + PowerMockito.doThrow(directBufferMemoryAllocationException) + .when(spyStorageEngine) + .buildNewDataRegion( + ArgumentMatchers.eq("root.sg"), ArgumentMatchers.any(DataRegionId.class)); + + List<Future<Void>> futures = new ArrayList<>(); + Method asyncRecoverMethod = StorageEngine.class.getDeclaredMethod("asyncRecover", List.class); + asyncRecoverMethod.setAccessible(true); + try { + asyncRecoverMethod.invoke(spyStorageEngine, futures); + + Assert.assertEquals(1, futures.size()); + try { + futures.get(0).get(); + Assert.fail("Expected data region recovery to fail."); + } catch (ExecutionException e) { + Assert.assertSame(directBufferMemoryAllocationException, e.getCause()); + } + + ExceptionalCountDownLatch latch = + WALRecoverManager.getInstance().getAllDataRegionScannedLatch(); + Assert.assertTrue(latch.hasException()); + Assert.assertEquals( + directBufferMemoryAllocationException.getMessage(), latch.getExceptionMessage()); + } finally { + WALRecoverManager.getInstance().clear(); + testThreadPool.shutdownNow(); + setCachedThreadPool(spyStorageEngine, null); + } + } + + @Test + public void testNotifyWALRecoverManagerButContinueForOtherDataRegionException() throws Exception { + StorageEngine spyStorageEngine = PowerMockito.spy(storageEngine); + DataRegionException dataRegionException = new DataRegionException("other recovery failure"); + ExecutorService testThreadPool = Executors.newSingleThreadExecutor(); + setCachedThreadPool(spyStorageEngine, testThreadPool); + PowerMockito.doReturn( + Collections.singletonMap("root.sg", Lists.newArrayList(new DataRegionId(0)))) + .when(spyStorageEngine) + .getLocalDataRegionInfo(); + PowerMockito.doThrow(dataRegionException) + .when(spyStorageEngine) + .buildNewDataRegion( + ArgumentMatchers.eq("root.sg"), ArgumentMatchers.any(DataRegionId.class)); + + List<Future<Void>> futures = new ArrayList<>(); + Method asyncRecoverMethod = StorageEngine.class.getDeclaredMethod("asyncRecover", List.class); + asyncRecoverMethod.setAccessible(true); + try { + asyncRecoverMethod.invoke(spyStorageEngine, futures); + + Assert.assertEquals(1, futures.size()); + futures.get(0).get(); + + ExceptionalCountDownLatch latch = + WALRecoverManager.getInstance().getAllDataRegionScannedLatch(); + Assert.assertTrue(latch.hasException()); + Assert.assertEquals(dataRegionException.getMessage(), latch.getExceptionMessage()); + } finally { + WALRecoverManager.getInstance().clear(); + testThreadPool.shutdownNow(); + setCachedThreadPool(spyStorageEngine, null); + } + } + + private void setCachedThreadPool(StorageEngine storageEngine, ExecutorService executorService) + throws Exception { + Field cachedThreadPoolField = StorageEngine.class.getDeclaredField("cachedThreadPool"); + cachedThreadPoolField.setAccessible(true); + cachedThreadPoolField.set(storageEngine, executorService); + } }
