qiaojialin commented on a change in pull request #4498: URL: https://github.com/apache/iotdb/pull/4498#discussion_r768798327
########## File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionExceptionHandler.java ########## @@ -0,0 +1,246 @@ +/* + * 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.engine.compaction.inner; + +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils; +import org.apache.iotdb.db.engine.modification.Modification; +import org.apache.iotdb.db.engine.modification.ModificationFile; +import org.apache.iotdb.db.engine.storagegroup.TsFileManager; +import org.apache.iotdb.db.engine.storagegroup.TsFileResource; +import org.apache.iotdb.db.engine.storagegroup.TsFileResourceList; +import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter; + +import org.apache.commons.io.FileUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +/** + * This class is used to handle exception (including OOM error) occurred during compaction. The + * <i>allowCompaction</i> flag in {@link org.apache.iotdb.db.engine.storagegroup.TsFileManager} may + * be set to false if exception cannot be handled correctly(such as OOM during handling exception), + * after which the subsequent compaction in this storage group will not be carried out. Under some + * serious circumstances(such as data lost), the system will be set to read-only. + */ +public class InnerSpaceCompactionExceptionHandler { + private static final Logger LOGGER = LoggerFactory.getLogger("COMPACTION"); + + public static void handleException( + String fullStorageGroupName, + File logFile, + TsFileResource targetTsFile, + List<TsFileResource> selectedTsFileResourceList, + TsFileManager tsFileManager, + TsFileResourceList tsFileResourceList) { + + if (!logFile.exists()) { + // log file does not exist + // it means the compaction has not started yet + // we need not to handle it + return; + } + + boolean handleSuccess = true; + + List<TsFileResource> lostSourceFiles = new ArrayList<>(); + boolean allSourceFileExist = + checkAllSourceFilesExist(selectedTsFileResourceList, lostSourceFiles); + + if (allSourceFileExist) { + handleSuccess = + handleWhenAllSourceFilesExist( + fullStorageGroupName, targetTsFile, selectedTsFileResourceList, tsFileResourceList); + } else { + // some source file does not exists + // it means we start to delete source file + LOGGER.info( + "{} [Compaction][ExceptionHandler] some source files {} is lost", + fullStorageGroupName, + lostSourceFiles); + if (!targetTsFile.getTsFile().exists()) { + // some source files are missed, and target file not exists + // some data is lost, set the system to read-only + LOGGER.warn( + "{} [Compaction][ExceptionHandler] target file {} does not exist either, do nothing. Set system to read-only", + fullStorageGroupName, + targetTsFile); + IoTDBDescriptor.getInstance().getConfig().setReadOnly(true); + handleSuccess = false; + } else { + handleSuccess = + handleWhenSomeSourceFilesLost( + fullStorageGroupName, + targetTsFile, + selectedTsFileResourceList, + tsFileResourceList, + lostSourceFiles); + } + } + + if (!handleSuccess) { + LOGGER.error( + "{} [Compaction][ExceptionHandler] Failed to handle exception, set allowCompaction to false", + fullStorageGroupName); + tsFileManager.setAllowCompaction(false); + } else { + LOGGER.info( + "{} [Compaction][ExceptionHandler] Handle exception successfully, delete log file {}", + fullStorageGroupName, + logFile); + try { + FileUtils.delete(logFile); + } catch (IOException e) { + LOGGER.error( + "{} [Compaction][ExceptionHandler] Exception occurs while deleting log file {}, set allowCompaction to false", + fullStorageGroupName, + logFile, + e); + tsFileManager.setAllowCompaction(false); + } + } + } + + private static boolean checkAllSourceFilesExist( + List<TsFileResource> sourceFiles, List<TsFileResource> lostSourceFiles) { + boolean allSourceFileExist = true; + for (TsFileResource sourceTsFile : sourceFiles) { + if (!sourceTsFile.getTsFile().exists()) { + allSourceFileExist = false; + lostSourceFiles.add(sourceTsFile); + } + } + return allSourceFileExist; + } + + private static boolean handleWhenAllSourceFilesExist( + String fullStorageGroupName, + TsFileResource targetTsFile, + List<TsFileResource> selectedTsFileResourceList, + TsFileResourceList tsFileResourceList) { + // all source file exists, delete the target file + LOGGER.info( + "{} [Compaction][ExceptionHandler] all source files {} exists, delete target file {}", + fullStorageGroupName, + selectedTsFileResourceList, + targetTsFile); + if (!targetTsFile.remove()) { + // failed to remove target tsfile + // system should not carry out the subsequent compaction in case of data redundant + LOGGER.warn( + "{} [Compaction][ExceptionHandler] failed to remove target file {}", + fullStorageGroupName, + targetTsFile); + return false; + } + // deal with compaction modification + try { + for (TsFileResource sourceFile : selectedTsFileResourceList) { + if (sourceFile.getCompactionModFile().exists()) { + ModificationFile compactionModificationFile = + ModificationFile.getCompactionMods(sourceFile); + Collection<Modification> newModification = compactionModificationFile.getModifications(); + compactionModificationFile.close(); + // write the modifications to a new modification file + sourceFile.resetModFile(); + try (ModificationFile newModificationFile = sourceFile.getModFile()) { + for (Modification modification : newModification) { + newModificationFile.write(modification); + } + } + FileUtils.delete(new File(ModificationFile.getCompactionMods(sourceFile).getFilePath())); + } + } + } catch (Throwable e) { + LOGGER.error( + "{} Exception occurs while handling exception, set allowCompaction to false", + fullStorageGroupName, + e); + return false; + } + return true; + } + + private static boolean handleWhenSomeSourceFilesLost( + String fullStorageGroupName, + TsFileResource targetTsFile, + List<TsFileResource> selectedTsFileResourceList, + TsFileResourceList tsFileResourceList, + List<TsFileResource> lostSourceFiles) { + boolean handleSuccess = true; + try { + RestorableTsFileIOWriter writer = new RestorableTsFileIOWriter(targetTsFile.getTsFile()); + writer.close(); + if (!writer.hasCrashed()) { Review comment: Use a method such as: TsFileUtils.isTsFileComplete() to replace this writer -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
