[NO ISSUE][STO] Ensure Files From Failed Bulkload Are Deleted - user model changes: no - storage format changes: no - interface changes: no
Details: - If a bulkload is interrupted at the end-stage (i.e. after all tuples have been added but before the component is added to the index's disk components), then the bulkload operation will fail but the files from the failed operation are not deleted. This change ensures that if a bulkload fails in such stage, the files of the generated component are deleted to avoid file is already mapped exception if the bulkload is attempted again. - Ensure disk component is only destroyed once in case of bulkload cleanup. - Stop attempting to write a checkpoint early if the failure was due to the thread being interrupted. Change-Id: I7f63054ac72a9482e779e49eb4da658a08fe7e9d Reviewed-on: https://asterix-gerrit.ics.uci.edu/2970 Reviewed-by: Murtadha Hubail <[email protected]> Sonar-Qube: Jenkins <[email protected]> Tested-by: Jenkins <[email protected]> Integration-Tests: Jenkins <[email protected]> Reviewed-by: Michael Blow <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/asterixdb/repo Commit: http://git-wip-us.apache.org/repos/asf/asterixdb/commit/3613cab7 Tree: http://git-wip-us.apache.org/repos/asf/asterixdb/tree/3613cab7 Diff: http://git-wip-us.apache.org/repos/asf/asterixdb/diff/3613cab7 Branch: refs/heads/master Commit: 3613cab7691f0257b4a8befa41eedcc55ee5b680 Parents: 8078e65 Author: Murtadha Hubail <[email protected]> Authored: Tue Sep 18 13:19:02 2018 +0300 Committer: Murtadha Hubail <[email protected]> Committed: Tue Sep 18 08:45:19 2018 -0700 ---------------------------------------------------------------------- .../asterix/app/nc/IndexCheckpointManager.java | 3 ++ .../ChainedLSMDiskComponentBulkLoader.java | 2 +- .../impls/LSMIndexDiskComponentBulkLoader.java | 44 +++++++++++++------- 3 files changed, 34 insertions(+), 15 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/asterixdb/blob/3613cab7/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java ---------------------------------------------------------------------- diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java index 3c62d99..7b08bad 100644 --- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java +++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java @@ -196,6 +196,9 @@ public class IndexCheckpointManager implements IIndexCheckpointManager { // ensure it was written correctly by reading it read(checkpointPath); return; + } catch (ClosedByInterruptException e) { + LOGGER.info("interrupted while writing checkpoint at {}", checkpointPath); + throw HyracksDataException.create(e); } catch (IOException e) { if (i == MAX_CHECKPOINT_WRITE_ATTEMPTS) { throw HyracksDataException.create(e); http://git-wip-us.apache.org/repos/asf/asterixdb/blob/3613cab7/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java ---------------------------------------------------------------------- diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java index ab59b59..3fa45c9 100644 --- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java +++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java @@ -101,8 +101,8 @@ public class ChainedLSMDiskComponentBulkLoader implements ILSMDiskComponentBulkL for (int i = 0; i < bulkloadersCount; i++) { bulkloaderChain.get(i).cleanupArtifacts();; } + diskComponent.deactivateAndDestroy(); } - diskComponent.deactivateAndDestroy(); } @Override http://git-wip-us.apache.org/repos/asf/asterixdb/blob/3613cab7/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexDiskComponentBulkLoader.java ---------------------------------------------------------------------- diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexDiskComponentBulkLoader.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexDiskComponentBulkLoader.java index 3a43ba7..c739ad0 100644 --- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexDiskComponentBulkLoader.java +++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexDiskComponentBulkLoader.java @@ -22,6 +22,7 @@ import org.apache.hyracks.api.exceptions.HyracksDataException; import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference; import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent; import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponentBulkLoader; +import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation; import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation.LSMIOOperationStatus; import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext; import org.apache.hyracks.storage.common.IIndexBulkLoader; @@ -31,6 +32,7 @@ public class LSMIndexDiskComponentBulkLoader implements IIndexBulkLoader { private final AbstractLSMIndex lsmIndex; private final ILSMDiskComponentBulkLoader componentBulkLoader; private final ILSMIndexOperationContext opCtx; + private boolean failed = false; public LSMIndexDiskComponentBulkLoader(AbstractLSMIndex lsmIndex, ILSMIndexOperationContext opCtx, float fillFactor, boolean verifyInput, long numElementsHint) throws HyracksDataException { @@ -68,20 +70,10 @@ public class LSMIndexDiskComponentBulkLoader implements IIndexBulkLoader { @Override public void end() throws HyracksDataException { try { - try { - lsmIndex.getIOOperationCallback().afterOperation(opCtx.getIoOperation()); - componentBulkLoader.end(); - } catch (Throwable th) { // NOSONAR Must not call afterFinalize without setting failure - opCtx.getIoOperation().setStatus(LSMIOOperationStatus.FAILURE); - opCtx.getIoOperation().setFailure(th); - throw th; - } finally { - lsmIndex.getIOOperationCallback().afterFinalize(opCtx.getIoOperation()); - } - if (opCtx.getIoOperation().getStatus() == LSMIOOperationStatus.SUCCESS - && opCtx.getIoOperation().getNewComponent().getComponentSize() > 0) { - lsmIndex.getHarness().addBulkLoadedComponent(opCtx.getIoOperation()); - } + presistComponentToDisk(); + } catch (Throwable th) { // NOSONAR must cleanup in case of any failure + fail(th); + throw th; } finally { lsmIndex.getIOOperationCallback().completed(opCtx.getIoOperation()); } @@ -116,4 +108,28 @@ public class LSMIndexDiskComponentBulkLoader implements IIndexBulkLoader { return opCtx.getIoOperation().getFailure(); } + private void presistComponentToDisk() throws HyracksDataException { + try { + lsmIndex.getIOOperationCallback().afterOperation(opCtx.getIoOperation()); + componentBulkLoader.end(); + } catch (Throwable th) { // NOSONAR Must not call afterFinalize without setting failure + fail(th); + throw th; + } finally { + lsmIndex.getIOOperationCallback().afterFinalize(opCtx.getIoOperation()); + } + if (opCtx.getIoOperation().getStatus() == LSMIOOperationStatus.SUCCESS + && opCtx.getIoOperation().getNewComponent().getComponentSize() > 0) { + lsmIndex.getHarness().addBulkLoadedComponent(opCtx.getIoOperation()); + } + } + + private void fail(Throwable th) { + if (!failed) { + failed = true; + final ILSMIOOperation loadOp = opCtx.getIoOperation(); + loadOp.setFailure(th); + lsmIndex.cleanUpFilesForFailedOperation(loadOp); + } + } } \ No newline at end of file
