You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2022/05/17 07:19:59 UTC
[iotdb] branch rel/0.13 updated: [IOTDB-3160] TsFile will be corrupted when flushing memtable appears OOM (#5926)
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch rel/0.13
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/0.13 by this push:
new 0762e1e897 [IOTDB-3160] TsFile will be corrupted when flushing memtable appears OOM (#5926)
0762e1e897 is described below
commit 0762e1e897f1894d2d51f7ca10d8a4d57d6335a9
Author: Alan Choo <43...@users.noreply.github.com>
AuthorDate: Tue May 17 15:19:53 2022 +0800
[IOTDB-3160] TsFile will be corrupted when flushing memtable appears OOM (#5926)
---
.../iotdb/db/integration/IoTDBRestartIT.java | 99 ++++++++++++++--------
.../db/engine/storagegroup/TsFileProcessor.java | 57 ++++++++++---
.../writelog/manager/MultiFileLogNodeManager.java | 12 +++
.../write/writer/RestorableTsFileIOWriter.java | 37 +-------
4 files changed, 123 insertions(+), 82 deletions(-)
diff --git a/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBRestartIT.java b/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBRestartIT.java
index 2b940cfe00..63cc8403c2 100644
--- a/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBRestartIT.java
+++ b/integration/src/test/java/org/apache/iotdb/db/integration/IoTDBRestartIT.java
@@ -21,12 +21,17 @@ package org.apache.iotdb.db.integration;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
+import org.apache.iotdb.db.engine.memtable.IMemTable;
+import org.apache.iotdb.db.engine.storagegroup.TsFileProcessor;
import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.metadata.path.PartialPath;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.itbase.category.LocalStandaloneTest;
import org.apache.iotdb.jdbc.Config;
+import org.junit.After;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.slf4j.Logger;
@@ -48,12 +53,20 @@ public class IoTDBRestartIT {
private final Logger logger = LoggerFactory.getLogger(IoTDBRestartIT.class);
- @Test
- public void testRestart()
- throws SQLException, ClassNotFoundException, IOException, StorageEngineException {
+ @Before
+ public void setUp() throws Exception {
EnvironmentUtils.envSetUp();
Class.forName(Config.JDBC_DRIVER_NAME);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvironmentUtils.cleanEnv();
+ }
+ @Test
+ public void testRestart()
+ throws SQLException, ClassNotFoundException, IOException, StorageEngineException {
try (Connection connection =
DriverManager.getConnection(
Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
@@ -99,16 +112,11 @@ public class IoTDBRestartIT {
}
}
}
-
- EnvironmentUtils.cleanEnv();
}
@Test
public void testRestartDelete()
throws SQLException, ClassNotFoundException, IOException, StorageEngineException {
- EnvironmentUtils.envSetUp();
- Class.forName(Config.JDBC_DRIVER_NAME);
-
try (Connection connection =
DriverManager.getConnection(
Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
@@ -169,16 +177,11 @@ public class IoTDBRestartIT {
resultSet.close();
}
}
-
- EnvironmentUtils.cleanEnv();
}
@Test
public void testRestartQueryLargerThanEndTime()
throws SQLException, ClassNotFoundException, IOException, StorageEngineException {
- EnvironmentUtils.envSetUp();
- Class.forName(Config.JDBC_DRIVER_NAME);
-
try (Connection connection =
DriverManager.getConnection(
Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
@@ -228,16 +231,11 @@ public class IoTDBRestartIT {
}
assertEquals(1, cnt);
}
-
- EnvironmentUtils.cleanEnv();
}
@Test
public void testRestartEndTime()
throws SQLException, ClassNotFoundException, IOException, StorageEngineException {
- EnvironmentUtils.envSetUp();
- Class.forName(Config.JDBC_DRIVER_NAME);
-
try (Connection connection =
DriverManager.getConnection(
Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
@@ -284,15 +282,10 @@ public class IoTDBRestartIT {
}
assertEquals(2, cnt);
}
-
- EnvironmentUtils.cleanEnv();
}
@Test
public void testRecoverWALMismatchDataType() throws Exception {
- EnvironmentUtils.envSetUp();
- Class.forName(Config.JDBC_DRIVER_NAME);
-
try (Connection connection =
DriverManager.getConnection(
Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
@@ -324,15 +317,10 @@ public class IoTDBRestartIT {
assertEquals(1, cnt);
}
}
-
- EnvironmentUtils.cleanEnv();
}
@Test
public void testRecoverWALDeleteSchema() throws Exception {
- EnvironmentUtils.envSetUp();
- Class.forName(Config.JDBC_DRIVER_NAME);
-
try (Connection connection =
DriverManager.getConnection(
Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
@@ -361,14 +349,10 @@ public class IoTDBRestartIT {
assertEquals(1, cnt);
}
}
-
- EnvironmentUtils.cleanEnv();
}
@Test
public void testRecoverWALDeleteSchemaCheckResourceTime() throws Exception {
- EnvironmentUtils.envSetUp();
- Class.forName(Config.JDBC_DRIVER_NAME);
IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
int avgSeriesPointNumberThreshold = config.getAvgSeriesPointNumberThreshold();
config.setAvgSeriesPointNumberThreshold(2);
@@ -411,6 +395,55 @@ public class IoTDBRestartIT {
config.setAvgSeriesPointNumberThreshold(avgSeriesPointNumberThreshold);
config.setSeqTsFileSize(tsFileSize);
config.setUnSeqTsFileSize(unFsFileSize);
- EnvironmentUtils.cleanEnv();
+ }
+
+ @Test
+ public void testRecoverFromFlushMemTableError() throws Exception {
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("insert into root.turbine1.d1(timestamp,s1,s2) values(1,1.1,2.2)");
+ }
+
+ // mock exception
+ TsFileProcessor[] tsFileProcessors =
+ StorageEngine.getInstance()
+ .getProcessor(new PartialPath("root.turbine1"))
+ .getWorkSequenceTsFileProcessors()
+ .toArray(new TsFileProcessor[0]);
+ Assert.assertEquals(1, tsFileProcessors.length);
+ IMemTable memTable = tsFileProcessors[0].getWorkMemTable();
+ memTable.clear();
+ memTable.addTextDataSize(1);
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("flush");
+ }
+
+ IoTDBDescriptor.getInstance().getConfig().setReadOnly(false);
+ EnvironmentUtils.restartDaemon();
+
+ try (Connection connection =
+ DriverManager.getConnection(
+ Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root", "root");
+ Statement statement = connection.createStatement()) {
+
+ boolean hasResultSet = statement.execute("select * from root.**");
+ assertTrue(hasResultSet);
+
+ try (ResultSet resultSet = statement.getResultSet()) {
+ int cnt = 0;
+ while (resultSet.next()) {
+ Assert.assertEquals("1", resultSet.getString(1));
+ Assert.assertEquals("1.1", resultSet.getString(2));
+ Assert.assertEquals("2.2", resultSet.getString(3));
+ cnt++;
+ }
+ Assert.assertEquals(1, cnt);
+ }
+ }
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index 71eaef12ef..187081f72d 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -1000,6 +1000,22 @@ public class TsFileProcessor {
}
}
+ /** This method will synchronize the memTable and release its flushing resources */
+ private void syncReleaseFlushedMemTable(IMemTable memTable) {
+ synchronized (memTable) {
+ releaseFlushedMemTable(memTable);
+ memTable.notifyAll();
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "{}: {} released a memtable (signal={}), flushingMemtables size ={}",
+ storageGroupName,
+ tsFileResource.getTsFile().getName(),
+ memTable.isSignalMemTable(),
+ flushingMemTables.size());
+ }
+ }
+ }
+
/**
* Take the first MemTable from the flushingMemTables and flush it. Called by a flush thread of
* the flush manager pool
@@ -1045,7 +1061,32 @@ public class TsFileProcessor {
tsFileResource.getTsFile().getName(),
e1);
}
- Thread.currentThread().interrupt();
+ // release resource
+ try {
+ syncReleaseFlushedMemTable(memTableToFlush);
+ // close wal node
+ MultiFileLogNodeManager.getInstance()
+ .closeNode(storageGroupName + "-" + tsFileResource.getTsFile().getName());
+ // make sure no query will search this file
+ tsFileResource.setTimeIndex(config.getTimeIndexLevel().getTimeIndex());
+ // this callback method will register this empty tsfile into TsFileManager
+ for (CloseFileListener closeFileListener : closeFileListeners) {
+ closeFileListener.onClosed(this);
+ }
+ // close writer
+ writer.close();
+ writer = null;
+ synchronized (flushingMemTables) {
+ flushingMemTables.notifyAll();
+ }
+ } catch (Exception e1) {
+ logger.error(
+ "{}: {} Release resource meets error",
+ storageGroupName,
+ tsFileResource.getTsFile().getName(),
+ e1);
+ }
+ return;
}
}
}
@@ -1087,19 +1128,9 @@ public class TsFileProcessor {
tsFileResource.getTsFile().getName(),
memTableToFlush.isSignalMemTable());
}
+
// for sync flush
- synchronized (memTableToFlush) {
- releaseFlushedMemTable(memTableToFlush);
- memTableToFlush.notifyAll();
- if (logger.isDebugEnabled()) {
- logger.debug(
- "{}: {} released a memtable (signal={}), flushingMemtables size ={}",
- storageGroupName,
- tsFileResource.getTsFile().getName(),
- memTableToFlush.isSignalMemTable(),
- flushingMemTables.size());
- }
- }
+ syncReleaseFlushedMemTable(memTableToFlush);
if (shouldClose && flushingMemTables.isEmpty() && writer != null) {
try {
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java b/server/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java
index 3a3dc08953..408d6fe52d 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java
@@ -140,6 +140,18 @@ public class MultiFileLogNodeManager implements WriteLogNodeManager, IService {
}
}
+ public void closeNode(String identifier) throws IOException {
+ WriteLogNode node = nodeMap.remove(identifier);
+ if (node != null) {
+ try {
+ node.close();
+ } catch (IOException e) {
+ logger.error("failed to close {}", node, e);
+ }
+ node.release();
+ }
+ }
+
@Override
public void close() {
logger.info("{} nodes to be closed", nodeMap.size());
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/RestorableTsFileIOWriter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/RestorableTsFileIOWriter.java
index 2ea48fafb8..70a5d8cf9f 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/RestorableTsFileIOWriter.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/RestorableTsFileIOWriter.java
@@ -75,42 +75,7 @@ public class RestorableTsFileIOWriter extends TsFileIOWriter {
* @throws IOException if write failed, or the file is broken but autoRepair==false.
*/
public RestorableTsFileIOWriter(File file) throws IOException {
- if (logger.isDebugEnabled()) {
- logger.debug("{} is opened.", file.getName());
- }
- this.file = file;
- this.out = FSFactoryProducer.getFileOutputFactory().getTsFileOutput(file.getPath(), true);
-
- // file doesn't exist
- if (file.length() == 0) {
- startFile();
- crashed = true;
- canWrite = true;
- return;
- }
-
- if (file.exists()) {
- try (TsFileSequenceReader reader = new TsFileSequenceReader(file.getAbsolutePath(), false)) {
-
- truncatedSize = reader.selfCheck(knownSchemas, chunkGroupMetadataList, true);
- minPlanIndex = reader.getMinPlanIndex();
- maxPlanIndex = reader.getMaxPlanIndex();
- if (truncatedSize == TsFileCheckStatus.COMPLETE_FILE) {
- crashed = false;
- canWrite = false;
- out.close();
- } else if (truncatedSize == TsFileCheckStatus.INCOMPATIBLE_FILE) {
- out.close();
- throw new NotCompatibleTsFileException(
- String.format("%s is not in TsFile format.", file.getAbsolutePath()));
- } else {
- crashed = true;
- canWrite = true;
- // remove broken data
- out.truncate(truncatedSize);
- }
- }
- }
+ this(file, true);
}
public RestorableTsFileIOWriter(File file, boolean truncate) throws IOException {