Merge "Merge branch 'trinity' into 'master'"
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
index a439fe5..d528925 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/NCAppRuntimeContext.java
@@ -255,6 +255,7 @@
datasetLifecycleManager =
new DatasetLifecycleManager(storageProperties, localResourceRepository, txnSubsystem.getLogManager(),
virtualBufferCache, indexCheckpointManagerProvider, ioManager.getIODevices().size());
+ localResourceRepository.setDatasetLifecycleManager(datasetLifecycleManager);
final String nodeId = getServiceContext().getNodeId();
final Set<Integer> nodePartitions = metadataProperties.getNodePartitions(nodeId);
replicaManager = new ReplicaManager(this, nodePartitions);
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
index 96ae699..39a8841 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
@@ -43,8 +43,10 @@
import java.util.function.Predicate;
import java.util.stream.Collectors;
+import org.apache.asterix.common.api.IDatasetLifecycleManager;
import org.apache.asterix.common.dataflow.DatasetLocalResource;
import org.apache.asterix.common.exceptions.AsterixException;
+import org.apache.asterix.common.replication.AllDatasetsReplicationStrategy;
import org.apache.asterix.common.replication.IReplicationManager;
import org.apache.asterix.common.replication.IReplicationStrategy;
import org.apache.asterix.common.replication.ReplicationJob;
@@ -106,6 +108,7 @@
private final IIndexCheckpointManagerProvider indexCheckpointManagerProvider;
private final IPersistedResourceRegistry persistedResourceRegistry;
private final ReentrantReadWriteLock resourcesAccessLock = new ReentrantReadWriteLock(true);
+ private IDatasetLifecycleManager datasetLifecycleManager;
public PersistentLocalResourceRepository(IIOManager ioManager,
IIndexCheckpointManagerProvider indexCheckpointManagerProvider,
@@ -391,6 +394,10 @@
}
}
+ public void setDatasetLifecycleManager(IDatasetLifecycleManager datasetLifecycleManager) {
+ this.datasetLifecycleManager = datasetLifecycleManager;
+ }
+
private void createReplicationJob(ReplicationOperation operation, FileReference fileRef)
throws HyracksDataException {
ReplicationJob job = new ReplicationJob(ReplicationJobType.METADATA, operation, ReplicationExecutionType.SYNC,
@@ -554,6 +561,7 @@
public void cleanup(int partition) throws HyracksDataException {
beforeReadAccess();
try {
+ datasetLifecycleManager.waitForIO(AllDatasetsReplicationStrategy.INSTANCE, partition);
final Set<FileReference> partitionIndexes = getPartitionIndexes(partition);
try {
for (FileReference index : partitionIndexes) {