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 4b027a8a64439..73469a53afad4 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 @@ -52,4 +52,6 @@ private DataNodeSchemaMessages() {} public static final String GET_TEMPLATE_INFO_ERROR = "Failed to get schema template [%s] from ConfigNode: %s"; public static final String ILLEGAL_PATH_LOG = "illegal path {}"; + public static final String SCHEMA_FILE_WRONG_VERSION = + "SchemaFile with wrong version, please check or upgrade."; } 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 7c94694932e65..8c1c011ec9ae3 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 @@ -45,4 +45,5 @@ private DataNodeSchemaMessages() {} public static final String GET_ALL_TEMPLATE_ERROR = "从 ConfigNode 获取元数据模板列表失败：%s"; public static final String GET_TEMPLATE_INFO_ERROR = "从 ConfigNode 获取元数据模板 [%s] 失败：%s"; public static final String ILLEGAL_PATH_LOG = "非法路径 {}"; + public static final String SCHEMA_FILE_WRONG_VERSION = "SchemaFile 版本错误，请检查或升级。"; } 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 1501f50acc1bc..a2df4a680cc53 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 @@ -29,6 +29,7 @@ import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.metadata.schemafile.SchemaFileNotExists; +import org.apache.iotdb.db.i18n.DataNodeSchemaMessages; import org.apache.iotdb.db.schemaengine.metric.SchemaRegionCachedMetric; 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; @@ -271,10 +272,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 @@ -353,28 +386,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("SchemaFile with wrong version, please check or upgrade."); + 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; + } } } @@ -389,7 +428,36 @@ private void updateHeaderBuffer() throws IOException { ReadWriteIOUtils.write(SchemaFileConfig.SCHEMA_FILE_VERSION, headerContent); headerContent.flip(); - channel.write(headerContent, 0); + IOUtils.writeFully(channel, headerContent, 0); + } + + 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..296521f525cdb 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 @@ -19,6 +19,7 @@ package org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile; import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.commons.utils.IOUtils; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.exception.metadata.schemafile.SegmentNotFoundException; @@ -106,7 +107,7 @@ public void syncPageBuffer() { @Override public void flushPageToChannel(FileChannel channel) throws IOException { this.pageBuffer.clear(); - channel.write(pageBuffer, SchemaFile.getPageAddress(pageIndex)); + IOUtils.writeFully(channel, pageBuffer, SchemaFile.getPageAddress(pageIndex)); 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 4e9d6524f1996..0fb8c0ce7f432 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 @@ -53,19 +53,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; } } @@ -75,9 +84,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; } } @@ -85,18 +111,30 @@ 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) { + // Check byte semantics with CRC32 or another method. + 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(); // complete log file if (!res.isEmpty()) { try (FileOutputStream outputStream = new FileOutputStream(logPath, true)) { - outputStream.write(new byte[] {SchemaFileConfig.SF_COMMIT_MARK}); + IOUtils.writeFully( + outputStream.getChannel(), + ByteBuffer.wrap(new byte[] {SchemaFileConfig.SF_COMMIT_MARK})); return outputStream.getChannel().size(); } } 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 56e0b5a6dc20a..9bca9e312961f 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 @@ -85,14 +85,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; } } @@ -525,7 +534,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 bb25ea10092c6..7bbfff69d90d1 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 @@ -20,12 +20,17 @@ import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.commons.i18n.UtilMessages; import org.apache.iotdb.commons.schema.node.IMNode; import org.apache.iotdb.commons.schema.node.role.IDatabaseMNode; 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.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; @@ -45,7 +53,13 @@ import org.junit.Before; import org.junit.Test; +import java.io.FileOutputStream; 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 +74,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 +111,248 @@ 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( + UtilMessages.EXCEPTION_FILE_CHANNEL_WRITE_MADE_NO_PROGRESS_B0F24ACD, + 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 = + ((FileOutputStream) getField(SchemaLogWriter.class, logWriter, "fileOutputStream")) + .getChannel(); + 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..037b7f4027550 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 @@ -19,10 +19,12 @@ package org.apache.iotdb.db.metadata.mtree.schemafile; import org.apache.iotdb.commons.exception.MetadataException; +import org.apache.iotdb.commons.i18n.UtilMessages; import org.apache.iotdb.commons.schema.node.utils.IMNodeFactory; 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,106 @@ 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( + UtilMessages.EXCEPTION_FILE_CHANNEL_WRITE_MADE_NO_PROGRESS_B0F24ACD, 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 = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/FileChannelWriteArchitectureTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/FileChannelWriteArchitectureTest.java new file mode 100644 index 0000000000000..2b11f5fbc6f50 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/FileChannelWriteArchitectureTest.java @@ -0,0 +1,91 @@ +/* + * 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.utils; + +import org.apache.iotdb.commons.utils.IOUtils; +import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.SchemaFile; + +import com.tngtech.archunit.base.DescribedPredicate; +import com.tngtech.archunit.core.domain.JavaMethodCall; +import com.tngtech.archunit.core.importer.ClassFileImporter; +import com.tngtech.archunit.core.importer.ImportOption; +import com.tngtech.archunit.lang.ArchRule; +import com.tngtech.archunit.lang.EvaluationResult; +import org.junit.Assert; +import org.junit.Test; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; + +import static com.tngtech.archunit.lang.syntax.ArchRuleDefinition.noClasses; + +public class FileChannelWriteArchitectureTest { + + private static final ArchRule RULE = + noClasses() + .should() + .callMethodWhere( + new DescribedPredicate("call FileChannel.write directly") { + @Override + public boolean test(JavaMethodCall call) { + return call.getTarget().getName().equals("write") + && call.getTarget().getOwner().isAssignableTo(FileChannel.class); + } + }) + .because("schema-file writes must handle short writes and fail on zero progress"); + + @Test + public void schemaFileProductionCodeMustWriteFully() { + RULE.check( + new ClassFileImporter() + .withImportOption(new ImportOption.DoNotIncludeTests()) + .importPackages(SchemaFile.class.getPackage().getName())); + } + + @Test + public void ruleRejectsEveryRawWriteOverload() { + EvaluationResult result = RULE.evaluate(new ClassFileImporter().importClasses(RawWrites.class)); + + Assert.assertTrue(result.hasViolation()); + Assert.assertEquals(4, result.getFailureReport().getDetails().size()); + } + + @Test + public void ruleAllowsWriteFully() { + RULE.check(new ClassFileImporter().importClasses(CompleteWrites.class)); + } + + private static class RawWrites { + void write(FileChannel channel, ByteBuffer buffer, ByteBuffer[] buffers) throws IOException { + channel.write(buffer); + channel.write(buffer, 0); + channel.write(buffers); + channel.write(buffers, 0, buffers.length); + } + } + + private static class CompleteWrites { + void write(FileChannel channel, ByteBuffer buffer) throws IOException { + IOUtils.writeFully(channel, buffer); + IOUtils.writeFully(channel, buffer, 0); + } + } +} diff --git a/iotdb-core/node-commons/pom.xml b/iotdb-core/node-commons/pom.xml index 38d7cec2a5890..d3a8b00301071 100644 --- a/iotdb-core/node-commons/pom.xml +++ b/iotdb-core/node-commons/pom.xml @@ -30,6 +30,7 @@ IoTDB: Core: Node Commons Common modules of the server and config node + en ${commons.test.skip} false ${commons.test.skip} @@ -193,6 +194,24 @@ + + org.codehaus.mojo + build-helper-maven-plugin + + + add-i18n-source + + add-source + + generate-sources + + + ${project.basedir}/src/main/i18n/${i18n.locale} + + + + + @@ -295,6 +314,12 @@ + + with-zh-locale + + zh + + skipCommonsTests diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/UtilMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/UtilMessages.java new file mode 100644 index 0000000000000..e4ea264564204 --- /dev/null +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/UtilMessages.java @@ -0,0 +1,27 @@ +/* + * 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.commons.i18n; + +public final class UtilMessages { + + private UtilMessages() {} + + public static final String EXCEPTION_FILE_CHANNEL_WRITE_MADE_NO_PROGRESS_B0F24ACD = + "File channel write made no progress"; +} diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/UtilMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/UtilMessages.java new file mode 100644 index 0000000000000..dd7fa81e2bd4c --- /dev/null +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/UtilMessages.java @@ -0,0 +1,27 @@ +/* + * 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.commons.i18n; + +public final class UtilMessages { + + private UtilMessages() {} + + public static final String EXCEPTION_FILE_CHANNEL_WRITE_MADE_NO_PROGRESS_B0F24ACD = + "文件通道写入未取得进展"; +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/IOUtils.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/IOUtils.java index 8d771f9947b20..cae779a9fa674 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/IOUtils.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/IOUtils.java @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.auth.entity.PriPrivilegeType; import org.apache.iotdb.commons.auth.entity.Role; import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.commons.i18n.UtilMessages; import org.apache.iotdb.commons.path.PartialPath; import com.google.common.base.Supplier; @@ -118,6 +119,42 @@ public static void readFully(FileChannel fileChannel, ByteBuffer buffer, long po } } + /** + * Writes all remaining bytes without changing the channel's current position. + * + * @return the number of bytes written + * @throws IOException if a write fails or makes no progress + */ + public static int writeFully(FileChannel fileChannel, ByteBuffer buffer, long position) + throws IOException { + final int size = buffer.remaining(); + while (buffer.hasRemaining()) { + final int written = fileChannel.write(buffer, position); + if (written <= 0) { + throw new IOException(UtilMessages.EXCEPTION_FILE_CHANNEL_WRITE_MADE_NO_PROGRESS_B0F24ACD); + } + position += written; + } + return size; + } + + /** + * Writes all remaining bytes at the channel's current position, advancing it as bytes are + * written. + * + * @return the number of bytes written + * @throws IOException if a write fails or makes no progress + */ + public static int writeFully(FileChannel fileChannel, ByteBuffer buffer) throws IOException { + final int size = buffer.remaining(); + while (buffer.hasRemaining()) { + if (fileChannel.write(buffer) <= 0) { + throw new IOException(UtilMessages.EXCEPTION_FILE_CHANNEL_WRITE_MADE_NO_PROGRESS_B0F24ACD); + } + } + return size; + } + /** * Read a string from the given stream. * diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/IOUtilsTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/IOUtilsTest.java index bf2cf546628a1..54edb62ae5c66 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/IOUtilsTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/IOUtilsTest.java @@ -18,6 +18,8 @@ */ package org.apache.iotdb.commons.utils; +import org.apache.iotdb.commons.i18n.UtilMessages; + import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; @@ -29,12 +31,162 @@ import java.io.InputStream; import java.nio.ByteBuffer; import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; import java.util.concurrent.atomic.AtomicInteger; public class IOUtilsTest { private static final String ENCODING = "UTF-8"; + @Test + public void writeFullyCompletesShortPositionalWrites() throws IOException { + ByteBuffer buffer = ByteBuffer.wrap(new byte[] {9, 1, 2, 3, 8}); + buffer.position(1); + buffer.limit(4); + ByteArrayOutputStream persisted = new ByteArrayOutputStream(); + FileChannel channel = Mockito.mock(FileChannel.class); + long start = Integer.MAX_VALUE + 5L; + Mockito.when(channel.write(Mockito.any(ByteBuffer.class), Mockito.anyLong())) + .thenAnswer( + invocation -> { + Assert.assertEquals(start + persisted.size(), (long) invocation.getArgument(1)); + persisted.write(((ByteBuffer) invocation.getArgument(0)).get()); + return 1; + }); + + Assert.assertEquals(3, IOUtils.writeFully(channel, buffer, start)); + + Assert.assertArrayEquals(new byte[] {1, 2, 3}, persisted.toByteArray()); + Assert.assertEquals(4, buffer.position()); + Assert.assertEquals(4, buffer.limit()); + } + + @Test + public void writeFullyCompletesShortSequentialWrites() throws IOException { + ByteBuffer buffer = ByteBuffer.wrap(new byte[] {1, 2, 3}); + ByteArrayOutputStream persisted = new ByteArrayOutputStream(); + FileChannel channel = Mockito.mock(FileChannel.class); + Mockito.when(channel.write(Mockito.any(ByteBuffer.class))) + .thenAnswer( + invocation -> { + persisted.write(((ByteBuffer) invocation.getArgument(0)).get()); + return 1; + }); + + Assert.assertEquals(3, IOUtils.writeFully(channel, buffer)); + + Assert.assertArrayEquals(new byte[] {1, 2, 3}, persisted.toByteArray()); + Assert.assertFalse(buffer.hasRemaining()); + } + + @Test + public void writeFullyDoesNotWriteEmptyBuffers() throws IOException { + FileChannel channel = Mockito.mock(FileChannel.class); + ByteBuffer buffer = ByteBuffer.allocate(0); + + Assert.assertEquals(0, IOUtils.writeFully(channel, buffer, 7)); + Assert.assertEquals(0, IOUtils.writeFully(channel, buffer)); + + Mockito.verifyZeroInteractions(channel); + } + + @Test + public void writeFullyFailsImmediatelyWhenNoProgressIsMade() throws IOException { + for (boolean positional : new boolean[] {false, true}) { + for (int noProgress : new int[] {0, -1}) { + FileChannel channel = Mockito.mock(FileChannel.class); + IOException unexpectedRetry = new IOException("Unexpected retry after no progress"); + if (positional) { + Mockito.when(channel.write(Mockito.any(ByteBuffer.class), Mockito.anyLong())) + .thenAnswer( + invocation -> { + ((ByteBuffer) invocation.getArgument(0)).get(); + return 1; + }) + .thenReturn(noProgress) + .thenThrow(unexpectedRetry); + } else { + Mockito.when(channel.write(Mockito.any(ByteBuffer.class))) + .thenAnswer( + invocation -> { + ((ByteBuffer) invocation.getArgument(0)).get(); + return 1; + }) + .thenReturn(noProgress) + .thenThrow(unexpectedRetry); + } + ByteBuffer buffer = ByteBuffer.wrap(new byte[] {1, 2, 3}); + + IOException failure = + Assert.assertThrows( + IOException.class, + () -> { + if (positional) { + IOUtils.writeFully(channel, buffer, 5); + } else { + IOUtils.writeFully(channel, buffer); + } + }); + + Assert.assertEquals( + UtilMessages.EXCEPTION_FILE_CHANNEL_WRITE_MADE_NO_PROGRESS_B0F24ACD, + failure.getMessage()); + Assert.assertEquals(1, buffer.position()); + if (positional) { + Mockito.verify(channel, Mockito.times(2)) + .write(Mockito.any(ByteBuffer.class), Mockito.anyLong()); + } else { + Mockito.verify(channel, Mockito.times(2)).write(Mockito.any(ByteBuffer.class)); + } + } + } + } + + @Test + public void writeFullyPropagatesOriginalIOException() throws IOException { + for (boolean positional : new boolean[] {false, true}) { + FileChannel channel = Mockito.mock(FileChannel.class); + IOException failure = new IOException("Injected write failure"); + if (positional) { + Mockito.when(channel.write(Mockito.any(ByteBuffer.class), Mockito.anyLong())) + .thenThrow(failure); + } else { + Mockito.when(channel.write(Mockito.any(ByteBuffer.class))).thenThrow(failure); + } + + Assert.assertSame( + failure, + Assert.assertThrows( + IOException.class, + () -> { + if (positional) { + IOUtils.writeFully(channel, ByteBuffer.allocate(1), 0); + } else { + IOUtils.writeFully(channel, ByteBuffer.allocate(1)); + } + })); + } + } + + @Test + public void writeFullyPreservesPositionalAndSequentialChannelSemantics() throws IOException { + Path path = Files.createTempFile("iotdb-write-fully", ".bin"); + try (FileChannel channel = + FileChannel.open(path, StandardOpenOption.READ, StandardOpenOption.WRITE)) { + channel.position(2); + + Assert.assertEquals(2, IOUtils.writeFully(channel, ByteBuffer.wrap(new byte[] {1, 2}), 0)); + Assert.assertEquals(2, channel.position()); + Assert.assertEquals(1, IOUtils.writeFully(channel, ByteBuffer.wrap(new byte[] {3}))); + Assert.assertEquals(3, channel.position()); + Assert.assertArrayEquals(new byte[] {1, 2, 3}, Files.readAllBytes(path)); + } finally { + Files.delete(path); + } + } + @Test public void readStringReadsCompletePayloadAfterShortRead() throws IOException { ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); diff --git a/pom.xml b/pom.xml index d6eb4ba7d972e..d72382c773237 100644 --- a/pom.xml +++ b/pom.xml @@ -151,7 +151,7 @@ 0.23.0 1.9 1.5.6-3 - 1.1.4-260130-SNAPSHOT + 1.1.5-SNAPSHOT