diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java index 7fde13b406338..102871ddd73e4 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java @@ -294,6 +294,11 @@ public final class DataNodeSchemaMessages { "PBTree File [{}] will be overwritten since already exists."; public static final String SCHEMA_FILE_WRONG_VERSION = "SchemaFile with wrong version, please check or upgrade."; + public static final String + EXCEPTION_SCHEMA_FILE_HEADER_WRITE_MADE_NO_PROGRESS_9AECE235 = + "Schema file header write made no progress"; + public static final String EXCEPTION_SCHEMA_PAGE_WRITE_MADE_NO_PROGRESS_E79C13A1 = + "Schema page write made no progress"; public static final String NODE_NO_CHILD_IN_PBTREE = "Node [%s] has no child in pbtree file."; public static final String SCHEMA_FILE_INSPECTED = "SchemaFile[%s] had been inspected."; public static final String FAILED_TO_CREATE_SCHEMA_FILE_SNAPSHOT = diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java index 26aadb5913926..cccc1d5258253 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java @@ -289,6 +289,11 @@ public final class DataNodeSchemaMessages { "PBTree 文件 [{}] 已存在，将被覆盖。"; public static final String SCHEMA_FILE_WRONG_VERSION = "SchemaFile 版本错误，请检查或升级。"; + public static final String + EXCEPTION_SCHEMA_FILE_HEADER_WRITE_MADE_NO_PROGRESS_9AECE235 = + "SchemaFile 头部写入没有进展"; + public static final String EXCEPTION_SCHEMA_PAGE_WRITE_MADE_NO_PROGRESS_E79C13A1 = + "SchemaPage 写入没有进展"; public static final String NODE_NO_CHILD_IN_PBTREE = "节点 [%s] 在 pbtree 文件中没有子节点。"; public static final String SCHEMA_FILE_INSPECTED = "SchemaFile[%s] 已被检查。"; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaFile.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaFile.java index 33a8df5adffac..9d37f2c915077 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaFile.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaFile.java @@ -273,10 +273,42 @@ public Iterator getChildren(ICachedMNode parent) @Override public void close() throws IOException { - updateHeaderBuffer(); - pageManager.close(); - forceChannel(); - channel.close(); + IOException failure = null; + try { + updateHeaderBuffer(); + } catch (IOException e) { + failure = e; + } + try { + pageManager.close(); + } catch (IOException e) { + if (failure == null) { + failure = e; + } else { + failure.addSuppressed(e); + } + } + try { + forceChannel(); + } catch (IOException e) { + if (failure == null) { + failure = e; + } else { + failure.addSuppressed(e); + } + } + try { + channel.close(); + } catch (IOException e) { + if (failure == null) { + failure = e; + } else { + failure.addSuppressed(e); + } + } + if (failure != null) { + throw failure; + } } @Override @@ -359,28 +391,34 @@ private void initFileHeader() throws IOException, MetadataException { if (channel.size() == 0) { // new pbtree file lastPageIndex = 0; - ReadWriteIOUtils.write(lastPageIndex, headerContent); - ReadWriteIOUtils.write(dataTTL, headerContent); - ReadWriteIOUtils.write(isEntity, headerContent); - ReadWriteIOUtils.write(sgNodeTemplateIdWithState, headerContent); - ReadWriteIOUtils.write(SchemaFileConfig.SCHEMA_FILE_VERSION, headerContent); lastSGAddr = 0L; - pageManager = new BTreePageManager(channel, pmtFile, -1, logPath); - } else { - IOUtils.readFully(channel, headerContent); - headerContent.clear(); - lastPageIndex = ReadWriteIOUtils.readInt(headerContent); - dataTTL = ReadWriteIOUtils.readLong(headerContent); - isEntity = ReadWriteIOUtils.readBool(headerContent); - sgNodeTemplateIdWithState = ReadWriteIOUtils.readInt(headerContent); - lastSGAddr = ReadWriteIOUtils.readLong(headerContent); - - if (ReadWriteIOUtils.readInt(headerContent) != SchemaFileConfig.SCHEMA_FILE_VERSION) { - channel.close(); - throw new MetadataException(DataNodeSchemaMessages.SCHEMA_FILE_WRONG_VERSION); + pageManager = null; + try { + pageManager = new BTreePageManager(channel, pmtFile, -1, logPath); + updateHeaderBuffer(); + } catch (IOException | MetadataException e) { + closeAfterInitializationFailure(e); + throw e; } + } else { + try { + IOUtils.readFully(channel, headerContent); + headerContent.clear(); + lastPageIndex = ReadWriteIOUtils.readInt(headerContent); + dataTTL = ReadWriteIOUtils.readLong(headerContent); + isEntity = ReadWriteIOUtils.readBool(headerContent); + sgNodeTemplateIdWithState = ReadWriteIOUtils.readInt(headerContent); + lastSGAddr = ReadWriteIOUtils.readLong(headerContent); + + if (ReadWriteIOUtils.readInt(headerContent) != SchemaFileConfig.SCHEMA_FILE_VERSION) { + throw new MetadataException(DataNodeSchemaMessages.SCHEMA_FILE_WRONG_VERSION); + } - pageManager = new BTreePageManager(channel, pmtFile, lastPageIndex, logPath); + pageManager = new BTreePageManager(channel, pmtFile, lastPageIndex, logPath); + } catch (IOException | MetadataException e) { + closeChannelAfterInitializationFailure(e); + throw e; + } } } @@ -395,7 +433,44 @@ private void updateHeaderBuffer() throws IOException { ReadWriteIOUtils.write(SchemaFileConfig.SCHEMA_FILE_VERSION, headerContent); headerContent.flip(); - channel.write(headerContent, 0); + long position = 0; + while (headerContent.hasRemaining()) { + int written = channel.write(headerContent, position); + if (written <= 0) { + throw new IOException( + DataNodeSchemaMessages.EXCEPTION_SCHEMA_FILE_HEADER_WRITE_MADE_NO_PROGRESS_9AECE235); + } + position += written; + } + } + + private void closeAfterInitializationFailure(Exception failure) { + if (pageManager != null) { + try { + pageManager.close(); + } catch (IOException closeException) { + failure.addSuppressed(closeException); + } + } + try { + channel.truncate(0); + } catch (IOException cleanupException) { + failure.addSuppressed(cleanupException); + } + try { + channel.force(true); + } catch (IOException cleanupException) { + failure.addSuppressed(cleanupException); + } + closeChannelAfterInitializationFailure(failure); + } + + private void closeChannelAfterInitializationFailure(Exception failure) { + try { + channel.close(); + } catch (IOException closeException) { + failure.addSuppressed(closeException); + } } private void forceChannel() throws IOException { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaPage.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaPage.java index 111dfaaa061df..db8efa705dc62 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaPage.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaPage.java @@ -21,6 +21,7 @@ import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.exception.metadata.schemafile.SegmentNotFoundException; +import org.apache.iotdb.db.i18n.DataNodeSchemaMessages; import org.apache.tsfile.utils.ReadWriteIOUtils; @@ -106,7 +107,15 @@ public void syncPageBuffer() { @Override public void flushPageToChannel(FileChannel channel) throws IOException { this.pageBuffer.clear(); - channel.write(pageBuffer, SchemaFile.getPageAddress(pageIndex)); + long position = SchemaFile.getPageAddress(pageIndex); + while (pageBuffer.hasRemaining()) { + int written = channel.write(pageBuffer, position); + if (written <= 0) { + throw new IOException( + DataNodeSchemaMessages.EXCEPTION_SCHEMA_PAGE_WRITE_MADE_NO_PROGRESS_E79C13A1); + } + position += written; + } dirtyFlag = false; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageIOChannel.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageIOChannel.java index f56e0c15d1b3c..ca7aa7764128c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageIOChannel.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageIOChannel.java @@ -52,19 +52,28 @@ public class PageIOChannel { this.channel = channel; this.pmtFile = pmtFile; this.readChannel = FileChannel.open(pmtFile.toPath(), StandardOpenOption.READ); - if (flushWithLogging) { - // without RATIS, utilize physical logging for integrity - int pageAcc = (int) recoverFromLog(logPath) / SchemaFileConfig.PAGE_LENGTH; - this.logWriter = new SchemaFileLogWriter(logPath); - logCounter = new AtomicInteger(pageAcc); - flushDirtyPagesStrategy = this::flushDirtyPagesWithLogging; - singlePageFlushStrategy = this::flushSinglePageWithLogging; - } else { - // with RATIS enabled, integrity is guaranteed by consensus protocol - logCounter = new AtomicInteger(); - logWriter = null; - flushDirtyPagesStrategy = this::flushDirtyPagesWithoutLogging; - singlePageFlushStrategy = this::flushSinglePageWithoutLogging; + try { + if (flushWithLogging) { + // without RATIS, utilize physical logging for integrity + int pageAcc = (int) recoverFromLog(logPath) / SchemaFileConfig.PAGE_LENGTH; + this.logWriter = new SchemaFileLogWriter(logPath); + logCounter = new AtomicInteger(pageAcc); + flushDirtyPagesStrategy = this::flushDirtyPagesWithLogging; + singlePageFlushStrategy = this::flushSinglePageWithLogging; + } else { + // with RATIS enabled, integrity is guaranteed by consensus protocol + logCounter = new AtomicInteger(); + logWriter = null; + flushDirtyPagesStrategy = this::flushDirtyPagesWithoutLogging; + singlePageFlushStrategy = this::flushSinglePageWithoutLogging; + } + } catch (IOException | MetadataException e) { + try { + close(); + } catch (IOException closeException) { + e.addSuppressed(closeException); + } + throw e; } } @@ -74,9 +83,26 @@ public void renewLogWriter() throws IOException { } } - public void closeLogWriter() throws IOException { + public void close() throws IOException { + IOException failure = null; if (logWriter != null) { - logWriter.close(); + try { + logWriter.close(); + } catch (IOException e) { + failure = e; + } + } + try { + readChannel.close(); + } catch (IOException e) { + if (failure == null) { + failure = e; + } else { + failure.addSuppressed(e); + } + } + if (failure != null) { + throw failure; } } @@ -84,11 +110,21 @@ public void closeLogWriter() throws IOException { private long recoverFromLog(String logPath) throws IOException, MetadataException { SchemaFileLogReader reader = new SchemaFileLogReader(logPath); ISchemaPage page; - List res = reader.collectUpdatedEntries(); - for (byte[] entry : res) { - // TODO check bytes semantic correctness with CRC32 or other way - page = ISchemaPage.loadSchemaPage(ByteBuffer.wrap(entry)); - page.flushPageToChannel(this.channel); + List res; + try { + res = reader.collectUpdatedEntries(); + for (byte[] entry : res) { + // TODO check bytes semantic correctness with CRC32 or other way + page = ISchemaPage.loadSchemaPage(ByteBuffer.wrap(entry)); + page.flushPageToChannel(this.channel); + } + } catch (IOException | MetadataException e) { + try { + reader.close(); + } catch (IOException closeException) { + e.addSuppressed(closeException); + } + throw e; } reader.close(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageManager.java index ea2f1f27f2810..3c0724760739f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageManager.java @@ -86,14 +86,23 @@ public abstract class PageManager implements IPageManager { pageIOChannel = new PageIOChannel(channel, pmtFile, true, logPath); } - // construct first page if file to init - if (lastPageIndex < 0) { - ISegmentedPage rootPage = - ISchemaPage.initSegmentedPage(ByteBuffer.allocate(SchemaFileConfig.PAGE_LENGTH), 0); - rootPage.allocNewSegment(SchemaFileConfig.SEG_MAX_SIZ); - pagePool.put(rootPage); - rootPage.syncPageBuffer(); - rootPage.flushPageToChannel(channel); + try { + // construct first page if file to init + if (lastPageIndex < 0) { + ISegmentedPage rootPage = + ISchemaPage.initSegmentedPage(ByteBuffer.allocate(SchemaFileConfig.PAGE_LENGTH), 0); + rootPage.allocNewSegment(SchemaFileConfig.SEG_MAX_SIZ); + pagePool.put(rootPage); + rootPage.syncPageBuffer(); + rootPage.flushPageToChannel(channel); + } + } catch (IOException | MetadataException e) { + try { + pageIOChannel.close(); + } catch (IOException closeException) { + e.addSuppressed(closeException); + } + throw e; } } @@ -528,7 +537,7 @@ public void inspect(PrintWriter pw) throws IOException, MetadataException { @Override public void close() throws IOException { - pageIOChannel.closeLogWriter(); + pageIOChannel.close(); } // endregion diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/mtree/schemafile/SchemaFileTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/mtree/schemafile/SchemaFileTest.java index 7813b0dd9a9a3..e0889a97204a3 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/mtree/schemafile/SchemaFileTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/mtree/schemafile/SchemaFileTest.java @@ -25,7 +25,12 @@ import org.apache.iotdb.commons.schema.node.role.IMeasurementMNode; import org.apache.iotdb.commons.schema.node.utils.IMNodeFactory; import org.apache.iotdb.commons.utils.PathUtils; +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.i18n.DataNodeSchemaMessages; import org.apache.iotdb.db.schemaengine.SchemaEngineMode; +import org.apache.iotdb.db.schemaengine.schemaregion.logfile.SchemaLogWriter; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.mnode.ICachedMNode; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.mnode.container.ICachedMNodeContainer; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.ISchemaFile; @@ -33,6 +38,9 @@ import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.SchemaFile; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.SchemaFileConfig; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.WrappedSegment; +import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.pagemgr.IPageManager; +import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.pagemgr.PageIOChannel; +import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.pagemgr.PageManager; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.loader.MNodeFactoryLoader; import org.apache.iotdb.db.schemaengine.schemaregion.utils.MNodeUtils; import org.apache.iotdb.db.utils.EnvironmentUtils; @@ -46,6 +54,11 @@ import org.junit.Test; import java.io.IOException; +import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; @@ -60,6 +73,13 @@ import java.util.stream.Collectors; import static org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.mnode.container.ICachedMNodeContainer.getCachedMNodeContainer; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.mockito.Mockito.withSettings; public class SchemaFileTest { @@ -90,6 +110,247 @@ private void writeMNodeInTest(ISchemaFile st, ICachedMNode node) st.writeMNode(node); } + @Test + public void testInitialHeaderHandlesShortWrites() throws Exception { + SchemaFile sf = SchemaFile.initSchemaFile("root.shortHeader", TEST_SCHEMA_REGION_ID); + sf.close(); + FileChannel channel = mock(FileChannel.class, withSettings().useConstructor()); + setField(SchemaFile.class, sf, "channel", channel); + ByteBuffer persistedHeader = ByteBuffer.allocate(29); + when(channel.write(any(ByteBuffer.class), anyLong())) + .thenAnswer( + invocation -> { + ByteBuffer source = invocation.getArgument(0); + long position = invocation.getArgument(1); + if (position >= SchemaFileConfig.FILE_HEADER_SIZE) { + int size = source.remaining(); + source.position(source.limit()); + return size; + } + int size = Math.min(3, source.remaining()); + persistedHeader.position((int) position); + for (int i = 0; i < size; i++) { + persistedHeader.put(source.get()); + } + return size; + }); + try { + initializeHeader(sf); + persistedHeader.flip(); + Assert.assertEquals(29, persistedHeader.remaining()); + Assert.assertEquals(0, persistedHeader.getInt()); + Assert.assertEquals(Long.MAX_VALUE, persistedHeader.getLong()); + Assert.assertEquals(0, persistedHeader.get()); + Assert.assertEquals(-1, persistedHeader.getInt()); + Assert.assertEquals(0L, persistedHeader.getLong()); + Assert.assertEquals(SchemaFileConfig.SCHEMA_FILE_VERSION, persistedHeader.getInt()); + } finally { + sf.close(); + } + } + + @Test + public void testInitialHeaderWriteWithNoProgressFails() throws Exception { + SchemaFile sf = SchemaFile.initSchemaFile("root.zeroHeaderWrite", TEST_SCHEMA_REGION_ID); + sf.close(); + FileChannel channel = mock(FileChannel.class, withSettings().useConstructor()); + setField(SchemaFile.class, sf, "channel", channel); + IOException fallbackFailure = new IOException("Header write was retried"); + int[] headerWriteCalls = {0}; + when(channel.write(any(ByteBuffer.class), anyLong())) + .thenAnswer( + invocation -> { + ByteBuffer source = invocation.getArgument(0); + long position = invocation.getArgument(1); + if (position >= SchemaFileConfig.FILE_HEADER_SIZE) { + int size = source.remaining(); + source.position(source.limit()); + return size; + } + headerWriteCalls[0]++; + if (headerWriteCalls[0] > 1) { + throw fallbackFailure; + } + return 0; + }); + + try { + initializeHeader(sf); + Assert.fail("Header write with no progress must abort initialization"); + } catch (InvocationTargetException e) { + Assert.assertEquals(1, headerWriteCalls[0]); + Assert.assertEquals( + DataNodeSchemaMessages.EXCEPTION_SCHEMA_FILE_HEADER_WRITE_MADE_NO_PROGRESS_9AECE235, + e.getCause().getMessage()); + } + } + + @Test + public void testClearPersistsHeaderForReopen() throws Exception { + SchemaFile sf = SchemaFile.initSchemaFile("root.clearedHeader", TEST_SCHEMA_REGION_ID); + sf.clear(); + + FileChannel channel = (FileChannel) getField(SchemaFile.class, sf, "channel"); + ByteBuffer persistedHeader = ByteBuffer.allocate(SchemaFileConfig.FILE_HEADER_SIZE); + long position = 0; + while (persistedHeader.hasRemaining()) { + int read = channel.read(persistedHeader, position); + Assert.assertTrue("Header must be present after clear", read > 0); + position += read; + } + persistedHeader.flip(); + Assert.assertEquals(0, persistedHeader.getInt()); + Assert.assertEquals(Long.MAX_VALUE, persistedHeader.getLong()); + Assert.assertEquals(0, persistedHeader.get()); + Assert.assertEquals(-1, persistedHeader.getInt()); + Assert.assertEquals(0L, persistedHeader.getLong()); + Assert.assertEquals(SchemaFileConfig.SCHEMA_FILE_VERSION, persistedHeader.getInt()); + + sf.close(); + + SchemaFile reopened = SchemaFile.loadSchemaFile("root.clearedHeader", TEST_SCHEMA_REGION_ID); + try { + Assert.assertNotNull(reopened.init()); + } finally { + reopened.close(); + } + } + + @Test + public void testInitialHeaderWriteFailureClosesResources() throws Exception { + IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + String originalProtocol = config.getSchemaRegionConsensusProtocolClass(); + config.setSchemaRegionConsensusProtocolClass(ConsensusFactory.SIMPLE_CONSENSUS); + try { + assertInitialHeaderWriteFailureClosesResources(); + } finally { + config.setSchemaRegionConsensusProtocolClass(originalProtocol); + } + } + + @Test + public void testInitialPageWriteFailureTruncatesFileAndClosesChannel() throws Exception { + IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + String originalProtocol = config.getSchemaRegionConsensusProtocolClass(); + config.setSchemaRegionConsensusProtocolClass(ConsensusFactory.SIMPLE_CONSENSUS); + try { + SchemaFile sf = SchemaFile.initSchemaFile("root.failedPage", TEST_SCHEMA_REGION_ID); + sf.close(); + FileChannel channel = mock(FileChannel.class, withSettings().useConstructor()); + setField(SchemaFile.class, sf, "channel", channel); + IOException failure = new IOException("Injected initial page write failure"); + when(channel.write(any(ByteBuffer.class), anyLong())) + .thenAnswer( + invocation -> { + long position = invocation.getArgument(1); + if (position >= SchemaFileConfig.FILE_HEADER_SIZE) { + throw failure; + } + ByteBuffer source = invocation.getArgument(0); + int size = source.remaining(); + source.position(source.limit()); + return size; + }); + + try { + initializeHeader(sf); + Assert.fail("Initial page write failure must abort initialization"); + } catch (InvocationTargetException e) { + Assert.assertSame(failure, e.getCause()); + } + verify(channel).truncate(0); + Assert.assertFalse(channel.isOpen()); + Assert.assertNull(getField(SchemaFile.class, sf, "pageManager")); + } finally { + config.setSchemaRegionConsensusProtocolClass(originalProtocol); + } + } + + private void assertInitialHeaderWriteFailureClosesResources() throws Exception { + SchemaFile sf = SchemaFile.initSchemaFile("root.failedHeader", TEST_SCHEMA_REGION_ID); + sf.close(); + FileChannel channel = mock(FileChannel.class, withSettings().useConstructor()); + setField(SchemaFile.class, sf, "channel", channel); + IOException failure = new IOException("Injected header write failure"); + when(channel.write(any(ByteBuffer.class), anyLong())) + .thenAnswer( + invocation -> { + ByteBuffer source = invocation.getArgument(0); + long position = invocation.getArgument(1); + if (position == 0) { + throw failure; + } + int size = source.remaining(); + source.position(source.limit()); + return size; + }); + try { + initializeHeader(sf); + Assert.fail("Header write failure must abort initialization"); + } catch (InvocationTargetException e) { + Assert.assertSame(failure, e.getCause()); + } + Object manager = getField(SchemaFile.class, sf, "pageManager"); + Object pageIO = getField(PageManager.class, manager, "pageIOChannel"); + FileChannel readChannel = (FileChannel) getField(PageIOChannel.class, pageIO, "readChannel"); + Object logWriter = getField(PageIOChannel.class, pageIO, "logWriter"); + FileChannel logChannel = + (FileChannel) getField(SchemaLogWriter.class, logWriter, "fileChannel"); + verify(channel).truncate(0); + Assert.assertFalse(channel.isOpen()); + Assert.assertFalse(readChannel.isOpen()); + Assert.assertFalse(logChannel.isOpen()); + } + + @Test + public void testCloseClosesChannelWhenPageManagerCloseFails() throws Exception { + SchemaFile sf = SchemaFile.initSchemaFile("root.closeFailure", TEST_SCHEMA_REGION_ID); + sf.close(); + FileChannel channel = mock(FileChannel.class, withSettings().useConstructor()); + IPageManager manager = mock(IPageManager.class); + setField(SchemaFile.class, sf, "channel", channel); + setField(SchemaFile.class, sf, "pageManager", manager); + when(manager.getLastPageIndex()).thenReturn(0); + when(channel.write(any(ByteBuffer.class), anyLong())) + .thenAnswer( + invocation -> { + ByteBuffer source = invocation.getArgument(0); + int size = source.remaining(); + source.position(source.limit()); + return size; + }); + IOException failure = new IOException("Injected page manager close failure"); + doThrow(failure).when(manager).close(); + + try { + sf.close(); + Assert.fail("Page manager close failure must be reported"); + } catch (IOException e) { + Assert.assertSame(failure, e); + } + verify(channel).force(true); + Assert.assertFalse(channel.isOpen()); + } + + private static void initializeHeader(SchemaFile sf) throws Exception { + Method initialize = SchemaFile.class.getDeclaredMethod("initFileHeader"); + initialize.setAccessible(true); + initialize.invoke(sf); + } + + private static Object getField(Class type, Object target, String name) throws Exception { + Field field = type.getDeclaredField(name); + field.setAccessible(true); + return field.get(target); + } + + private static void setField(Class type, Object target, String name, Object value) + throws Exception { + Field field = type.getDeclaredField(name); + field.setAccessible(true); + field.set(target, value); + } + @Test public void essentialTestSchemaFile() throws IOException, MetadataException { ISchemaFile sf = SchemaFile.initSchemaFile("root.test.vRoot1", TEST_SCHEMA_REGION_ID); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/mtree/schemafile/SchemaPageTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/mtree/schemafile/SchemaPageTest.java index 6206ff66ef22b..58450d17955ba 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/mtree/schemafile/SchemaPageTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/metadata/mtree/schemafile/SchemaPageTest.java @@ -20,9 +20,11 @@ import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.schema.node.utils.IMNodeFactory; +import org.apache.iotdb.db.i18n.DataNodeSchemaMessages; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.mnode.ICachedMNode; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.ISchemaPage; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.RecordUtils; +import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.SchemaFile; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.SchemaFileConfig; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.SchemaPage; import org.apache.iotdb.db.schemaengine.schemaregion.mtree.loader.MNodeFactoryLoader; @@ -38,6 +40,14 @@ import java.io.IOException; import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; public class SchemaPageTest { @@ -54,6 +64,107 @@ public void tearDown() throws Exception { EnvironmentUtils.cleanEnv(); } + @Test + public void testFlushPageHandlesShortWrites() throws Exception { + ByteBuffer buffer = ByteBuffer.allocate(SchemaFileConfig.PAGE_LENGTH); + for (int i = 0; i < buffer.capacity(); i++) { + buffer.put((byte) i); + } + ISchemaPage page = ISchemaPage.initSegmentedPage(buffer, 3); + page.syncPageBuffer(); + page.setDirtyFlag(); + byte[] expected = buffer.array().clone(); + ByteBuffer persisted = ByteBuffer.allocate(buffer.capacity()); + FileChannel channel = mock(FileChannel.class); + when(channel.write(any(ByteBuffer.class), anyLong())) + .thenAnswer( + invocation -> { + ByteBuffer source = invocation.getArgument(0); + long position = invocation.getArgument(1); + Assert.assertEquals(SchemaFile.getPageAddress(3) + persisted.position(), position); + Assert.assertTrue(page.isDirtyPage()); + int written = Math.min(4093, source.remaining()); + for (int i = 0; i < written; i++) { + persisted.put(source.get()); + } + return written; + }); + + page.flushPageToChannel(channel); + + Assert.assertEquals(buffer.capacity(), persisted.position()); + Assert.assertArrayEquals(expected, persisted.array()); + Assert.assertFalse(page.isDirtyPage()); + } + + @Test + public void testFlushPageWithNoProgressKeepsDirtyFlag() throws Exception { + ISchemaPage page = + ISchemaPage.initSegmentedPage(ByteBuffer.allocate(SchemaFileConfig.PAGE_LENGTH), 1); + page.setDirtyFlag(); + FileChannel channel = mock(FileChannel.class); + when(channel.write(any(ByteBuffer.class), anyLong())) + .thenAnswer( + invocation -> { + ByteBuffer source = invocation.getArgument(0); + source.position(source.position() + 3); + return 3; + }) + .thenReturn(0) + .thenThrow(new IOException("Page write was retried after making no progress")); + + try { + page.flushPageToChannel(channel); + Assert.fail("Page write with no progress must fail"); + } catch (IOException e) { + Assert.assertEquals( + DataNodeSchemaMessages.EXCEPTION_SCHEMA_PAGE_WRITE_MADE_NO_PROGRESS_E79C13A1, + e.getMessage()); + } + verify(channel, times(2)).write(any(ByteBuffer.class), anyLong()); + Assert.assertTrue(page.isDirtyPage()); + } + + @Test + public void testFlushPageRetriesWholePageAfterWriteFailure() throws Exception { + ISchemaPage page = + ISchemaPage.initSegmentedPage(ByteBuffer.allocate(SchemaFileConfig.PAGE_LENGTH), 2); + page.setDirtyFlag(); + FileChannel channel = mock(FileChannel.class); + IOException failure = new IOException("Injected page write failure"); + when(channel.write(any(ByteBuffer.class), anyLong())) + .thenAnswer( + invocation -> { + ByteBuffer source = invocation.getArgument(0); + source.position(source.position() + 3); + return 3; + }) + .thenThrow(failure) + .thenAnswer( + invocation -> { + ByteBuffer source = invocation.getArgument(0); + Assert.assertEquals(SchemaFile.getPageAddress(2), (long) invocation.getArgument(1)); + Assert.assertEquals(SchemaFileConfig.PAGE_LENGTH, source.remaining()); + Assert.assertTrue(page.isDirtyPage()); + int written = source.remaining(); + source.position(source.limit()); + return written; + }); + + try { + page.flushPageToChannel(channel); + Assert.fail("Page write failure must be reported"); + } catch (IOException e) { + Assert.assertSame(failure, e); + } + Assert.assertTrue(page.isDirtyPage()); + + page.flushPageToChannel(channel); + + verify(channel, times(3)).write(any(ByteBuffer.class), anyLong()); + Assert.assertFalse(page.isDirtyPage()); + } + @Test public void flatTreeInsert() throws IOException, MetadataException { ISchemaPage page =