[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 <mhubail@apache.org>
Sonar-Qube: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Integration-Tests: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Reviewed-by: Michael Blow <mblow@apache.org>
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 @@
// 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);
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 @@
for (int i = 0; i < bulkloadersCount; i++) {
bulkloaderChain.get(i).cleanupArtifacts();;
}
+ diskComponent.deactivateAndDestroy();
}
- diskComponent.deactivateAndDestroy();
}
@Override
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.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 @@
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 @@
@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 @@
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