You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by mm...@apache.org on 2021/09/02 23:22:12 UTC
[ignite] branch master updated: IGNITE-15413 Extract cancellation
token to rebuild index method (#9363)
This is an automated email from the ASF dual-hosted git repository.
mmuzaf pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 94ecb91 IGNITE-15413 Extract cancellation token to rebuild index method (#9363)
94ecb91 is described below
commit 94ecb9194962cf5f6f411dd79580a203cafd150a
Author: Maxim Muzafarov <mm...@apache.org>
AuthorDate: Fri Sep 3 02:21:43 2021 +0300
IGNITE-15413 Extract cancellation token to rebuild index method (#9363)
---
.../GridCommandHandlerIndexForceRebuildTest.java | 6 +++---
.../GridCommandHandlerIndexRebuildStatusTest.java | 4 ++--
.../internal/cache/query/index/IndexProcessor.java | 9 ++++++--
.../managers/indexing/IndexesRebuildTask.java | 19 ++++++++++-------
.../processors/query/GridQueryProcessor.java | 16 +++++++++------
...tionToken.java => IndexRebuildCancelToken.java} | 24 +++++++++++++---------
.../query/schema/SchemaIndexCacheFuture.java | 6 +++---
.../schema/SchemaIndexCachePartitionWorker.java | 20 ++++++++++--------
.../query/schema/SchemaIndexCacheVisitorImpl.java | 13 ++++++------
.../SchemaIndexOperationCancellationException.java | 2 +-
.../query/schema/SchemaOperationWorker.java | 6 +++---
.../cache/index/AbstractIndexingCommonTest.java | 4 ++--
.../DynamicEnableIndexingConcurrentSelfTest.java | 4 ++--
.../cache/index/IndexesRebuildTaskEx.java | 12 +++++++----
.../cache/index/StopRebuildIndexTest.java | 12 +++++------
.../IgnitePdsIndexingDefragmentationTest.java | 9 ++++++--
.../query/h2/GridIndexRebuildSelfTest.java | 4 ++--
17 files changed, 100 insertions(+), 70 deletions(-)
diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerIndexForceRebuildTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerIndexForceRebuildTest.java
index 80de9ab..750464d 100644
--- a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerIndexForceRebuildTest.java
+++ b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerIndexForceRebuildTest.java
@@ -34,9 +34,9 @@ import org.apache.ignite.internal.cache.query.index.IndexProcessor;
import org.apache.ignite.internal.managers.indexing.IndexesRebuildTask;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.IgniteCacheProxy;
+import org.apache.ignite.internal.processors.query.schema.IndexRebuildCancelToken;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheFuture;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorClosure;
-import org.apache.ignite.internal.processors.query.schema.SchemaIndexOperationCancellationToken;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.CU;
@@ -432,7 +432,7 @@ public class GridCommandHandlerIndexForceRebuildTest extends GridCommandHandlerA
SchemaIndexCacheFuture intlRebIdxFut = schemaIndexCacheFuture(n, CU.cacheId(cacheName2));
assertNotNull(intlRebIdxFut);
- assertTrue(waitForCondition(intlRebIdxFut.cancelToken()::isCancelled, getTestTimeout()));
+ assertTrue(waitForCondition(() -> intlRebIdxFut.cancelToken().cancelException() != null, getTestTimeout()));
stopLoad.set(true);
@@ -706,7 +706,7 @@ public class GridCommandHandlerIndexForceRebuildTest extends GridCommandHandlerA
private static class BlockingIndexesRebuildTask extends IndexesRebuildTask {
/** {@inheritDoc} */
@Override protected void startRebuild(GridCacheContext cctx, GridFutureAdapter<Void> fut,
- SchemaIndexCacheVisitorClosure clo, SchemaIndexOperationCancellationToken cancel) {
+ SchemaIndexCacheVisitorClosure clo, IndexRebuildCancelToken cancel) {
super.startRebuild(cctx, new BlockingRebuildIdxFuture(fut, cctx), clo, cancel);
}
}
diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerIndexRebuildStatusTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerIndexRebuildStatusTest.java
index aed6f5a..f128e5b 100644
--- a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerIndexRebuildStatusTest.java
+++ b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerIndexRebuildStatusTest.java
@@ -33,8 +33,8 @@ import org.apache.ignite.internal.commandline.CommandHandler;
import org.apache.ignite.internal.managers.indexing.IndexesRebuildTask;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
+import org.apache.ignite.internal.processors.query.schema.IndexRebuildCancelToken;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorClosure;
-import org.apache.ignite.internal.processors.query.schema.SchemaIndexOperationCancellationToken;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.visor.cache.index.IndexRebuildStatusInfoContainer;
@@ -243,7 +243,7 @@ public class GridCommandHandlerIndexRebuildStatusTest extends GridCommandHandler
*/
private static class BlockingIndexesRebuildTask extends IndexesRebuildTask {
@Override protected void startRebuild(GridCacheContext cctx, GridFutureAdapter<Void> fut,
- SchemaIndexCacheVisitorClosure clo, SchemaIndexOperationCancellationToken cancel) {
+ SchemaIndexCacheVisitorClosure clo, IndexRebuildCancelToken cancel) {
idxRebuildsStartedNum.incrementAndGet();
fut.listen((CI1<IgniteInternalFuture<?>>)f -> idxRebuildsStartedNum.decrementAndGet());
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/IndexProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/IndexProcessor.java
index 08fa6b5..fec4242 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/IndexProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/cache/query/index/IndexProcessor.java
@@ -60,6 +60,7 @@ import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
+import org.apache.ignite.internal.processors.query.schema.IndexRebuildCancelToken;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitor;
import org.apache.ignite.internal.util.GridAtomicLong;
import org.apache.ignite.internal.util.collection.IntMap;
@@ -345,8 +346,12 @@ public class IndexProcessor extends GridProcessorAdapter {
* @param force Force rebuild indexes.
* @return A future of rebuilding cache indexes.
*/
- @Nullable public IgniteInternalFuture<?> rebuildIndexesForCache(GridCacheContext<?, ?> cctx, boolean force) {
- return idxRebuild.rebuild(cctx, force);
+ @Nullable public IgniteInternalFuture<?> rebuildIndexesForCache(
+ GridCacheContext<?, ?> cctx,
+ boolean force,
+ IndexRebuildCancelToken cancelTok
+ ) {
+ return idxRebuild.rebuild(cctx, force, cancelTok);
}
/** */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/indexing/IndexesRebuildTask.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/indexing/IndexesRebuildTask.java
index 69a5d95..7d27bf3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/indexing/IndexesRebuildTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/indexing/IndexesRebuildTask.java
@@ -27,11 +27,11 @@ import org.apache.ignite.internal.cache.query.index.sorted.inline.InlineIndex;
import org.apache.ignite.internal.pagemem.store.IgnitePageStoreManager;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheContextInfo;
+import org.apache.ignite.internal.processors.query.schema.IndexRebuildCancelToken;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheFuture;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorClosure;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorImpl;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexOperationCancellationException;
-import org.apache.ignite.internal.processors.query.schema.SchemaIndexOperationCancellationToken;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.CU;
@@ -52,7 +52,11 @@ public class IndexesRebuildTask {
* @param force Force rebuild indexes.
* @return A future of rebuilding cache indexes.
*/
- @Nullable public IgniteInternalFuture<?> rebuild(GridCacheContext cctx, boolean force) {
+ @Nullable public IgniteInternalFuture<?> rebuild(
+ GridCacheContext<?, ?> cctx,
+ boolean force,
+ IndexRebuildCancelToken cancelTok
+ ) {
assert cctx != null;
if (!CU.affinityNode(cctx.localNode(), cctx.config().getNodeFilter()))
@@ -90,7 +94,7 @@ public class IndexesRebuildTask {
IgniteLogger log = cctx.kernalContext().grid().log();
// An internal future for the ability to cancel index rebuilding.
- SchemaIndexCacheFuture intRebFut = new SchemaIndexCacheFuture(new SchemaIndexOperationCancellationToken());
+ SchemaIndexCacheFuture intRebFut = new SchemaIndexCacheFuture(cancelTok);
SchemaIndexCacheFuture prevIntRebFut = idxRebuildFuts.put(cctx.cacheId(), intRebFut);
@@ -133,15 +137,15 @@ public class IndexesRebuildTask {
* @param cctx Cache context.
* @param fut Future for rebuild indexes.
* @param clo Closure.
- * @param cancel Cancellation token.
+ * @param cancelTok Cancellation token.
*/
protected void startRebuild(
GridCacheContext cctx,
GridFutureAdapter<Void> fut,
SchemaIndexCacheVisitorClosure clo,
- SchemaIndexOperationCancellationToken cancel
+ IndexRebuildCancelToken cancelTok
) {
- new SchemaIndexCacheVisitorImpl(cctx, cancel, fut).visit(clo);
+ new SchemaIndexCacheVisitorImpl(cctx, cancelTok, fut).visit(clo);
}
/**
@@ -161,7 +165,8 @@ public class IndexesRebuildTask {
* @param log Logger.
*/
private void cancelIndexRebuildFuture(@Nullable SchemaIndexCacheFuture rebFut, IgniteLogger log) {
- if (rebFut != null && !rebFut.isDone() && rebFut.cancelToken().cancel()) {
+ if (rebFut != null && !rebFut.isDone() &&
+ rebFut.cancelToken().cancel(new SchemaIndexOperationCancellationException("Index rebuild was cancelled."))) {
try {
rebFut.get();
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
index ba6e808..0bb70eb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
@@ -90,10 +90,10 @@ import org.apache.ignite.internal.processors.platform.PlatformProcessor;
import org.apache.ignite.internal.processors.query.aware.IndexBuildStatusStorage;
import org.apache.ignite.internal.processors.query.aware.IndexRebuildFutureStorage;
import org.apache.ignite.internal.processors.query.property.QueryBinaryProperty;
+import org.apache.ignite.internal.processors.query.schema.IndexRebuildCancelToken;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitor;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorClosure;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorImpl;
-import org.apache.ignite.internal.processors.query.schema.SchemaIndexOperationCancellationToken;
import org.apache.ignite.internal.processors.query.schema.SchemaOperationClientFuture;
import org.apache.ignite.internal.processors.query.schema.SchemaOperationException;
import org.apache.ignite.internal.processors.query.schema.SchemaOperationManager;
@@ -1856,7 +1856,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
* @throws SchemaOperationException If failed.
*/
public void processSchemaOperationLocal(SchemaAbstractOperation op, QueryTypeDescriptorImpl type, IgniteUuid depId,
- SchemaIndexOperationCancellationToken cancelTok) throws SchemaOperationException {
+ IndexRebuildCancelToken cancelTok) throws SchemaOperationException {
if (log.isDebugEnabled())
log.debug("Started local index operation [opId=" + op.id() + ']');
@@ -1969,7 +1969,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
}
if (idxRebuildFutStorage.prepareRebuildIndexes(singleton(cacheInfo.cacheId()), null).isEmpty())
- rebuildIndexesFromHash0(cacheInfo.cacheContext(), false);
+ rebuildIndexesFromHash0(cacheInfo.cacheContext(), false, cancelTok);
else {
if (log.isInfoEnabled())
log.info("Rebuilding indexes for the cache is already in progress: " + cacheInfo.name());
@@ -2368,7 +2368,7 @@ public class GridQueryProcessor extends GridProcessorAdapter {
}
try {
- return rebuildIndexesFromHash0(cctx, force);
+ return rebuildIndexesFromHash0(cctx, force, new IndexRebuildCancelToken());
}
finally {
busyLock.leaveBusy();
@@ -2381,8 +2381,12 @@ public class GridQueryProcessor extends GridProcessorAdapter {
* @param cctx Cache context.
* @param force Force rebuild indexes.
*/
- private IgniteInternalFuture<?> rebuildIndexesFromHash0(GridCacheContext<?, ?> cctx, boolean force) {
- IgniteInternalFuture<?> idxFut = idxProc.rebuildIndexesForCache(cctx, force);
+ private IgniteInternalFuture<?> rebuildIndexesFromHash0(
+ GridCacheContext<?, ?> cctx,
+ boolean force,
+ IndexRebuildCancelToken cancelTok
+ ) {
+ IgniteInternalFuture<?> idxFut = idxProc.rebuildIndexesForCache(cctx, force, cancelTok);
return chainIndexRebuildFuture(idxFut, cctx);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexOperationCancellationToken.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/IndexRebuildCancelToken.java
similarity index 68%
rename from modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexOperationCancellationToken.java
rename to modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/IndexRebuildCancelToken.java
index b74568c..a19bc14 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexOperationCancellationToken.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/IndexRebuildCancelToken.java
@@ -17,22 +17,26 @@
package org.apache.ignite.internal.processors.query.schema;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.Nullable;
/**
* Index operation cancellation token.
*/
-public class SchemaIndexOperationCancellationToken {
+public class IndexRebuildCancelToken {
/** Cancel flag. */
- private final AtomicBoolean flag = new AtomicBoolean();
+ private final AtomicReference<Throwable> flag;
+
+ /** Default constructor. */
+ public IndexRebuildCancelToken() {
+ flag = new AtomicReference<>();
+ }
/**
- * Get cancel state.
- *
- * @return {@code True} if cancelled.
+ * @return Exception that causes cancellation action.
*/
- public boolean isCancelled() {
+ public @Nullable Throwable cancelException() {
return flag.get();
}
@@ -41,12 +45,12 @@ public class SchemaIndexOperationCancellationToken {
*
* @return {@code True} if cancel flag was set by this call.
*/
- public boolean cancel() {
- return flag.compareAndSet(false, true);
+ public boolean cancel(Throwable e) {
+ return flag.compareAndSet(null, e);
}
/** {@inheritDoc} */
@Override public String toString() {
- return S.toString(SchemaIndexOperationCancellationToken.class, this);
+ return S.toString(IndexRebuildCancelToken.class, this);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexCacheFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexCacheFuture.java
index 28cfb5c..88ced58 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexCacheFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexCacheFuture.java
@@ -24,14 +24,14 @@ import org.apache.ignite.internal.util.future.GridFutureAdapter;
*/
public class SchemaIndexCacheFuture extends GridFutureAdapter<Void> {
/** Token for canceling index rebuilding. */
- private final SchemaIndexOperationCancellationToken cancelTok;
+ private final IndexRebuildCancelToken cancelTok;
/**
* Constructor.
*
* @param cancelTok Token for canceling index rebuilding.
*/
- public SchemaIndexCacheFuture(SchemaIndexOperationCancellationToken cancelTok) {
+ public SchemaIndexCacheFuture(IndexRebuildCancelToken cancelTok) {
this.cancelTok = cancelTok;
}
@@ -40,7 +40,7 @@ public class SchemaIndexCacheFuture extends GridFutureAdapter<Void> {
*
* @return Cancellation token.
*/
- public SchemaIndexOperationCancellationToken cancelToken() {
+ public IndexRebuildCancelToken cancelToken() {
return cancelTok;
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexCachePartitionWorker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexCachePartitionWorker.java
index c8ed9c4..e37fa54 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexCachePartitionWorker.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexCachePartitionWorker.java
@@ -67,7 +67,7 @@ public class SchemaIndexCachePartitionWorker extends GridWorker {
private final AtomicBoolean stop;
/** Cancellation token between all workers for all caches. */
- @Nullable private final SchemaIndexOperationCancellationToken cancel;
+ private final IndexRebuildCancelToken cancelTok;
/** Index closure. */
private final SchemaIndexCacheVisitorClosureWrapper wrappedClo;
@@ -87,7 +87,7 @@ public class SchemaIndexCachePartitionWorker extends GridWorker {
* @param cctx Cache context.
* @param locPart Partition.
* @param stop Stop flag between all workers for one cache.
- * @param cancel Cancellation token between all workers for all caches.
+ * @param cancelTok Cancellation token between all workers for all caches.
* @param clo Index closure.
* @param fut Worker future.
* @param partsCnt Count of partitions to be processed.
@@ -96,7 +96,7 @@ public class SchemaIndexCachePartitionWorker extends GridWorker {
GridCacheContext cctx,
GridDhtLocalPartition locPart,
AtomicBoolean stop,
- @Nullable SchemaIndexOperationCancellationToken cancel,
+ IndexRebuildCancelToken cancelTok,
SchemaIndexCacheVisitorClosure clo,
GridFutureAdapter<SchemaIndexCacheStat> fut,
AtomicInteger partsCnt
@@ -109,7 +109,7 @@ public class SchemaIndexCachePartitionWorker extends GridWorker {
this.cctx = cctx;
this.locPart = locPart;
- this.cancel = cancel;
+ this.cancelTok = cancelTok;
assert nonNull(stop);
assert nonNull(clo);
@@ -253,11 +253,15 @@ public class SchemaIndexCachePartitionWorker extends GridWorker {
/**
* Check if visit process is not cancelled.
*
- * @throws SchemaIndexOperationCancellationException If cancelled.
+ * @throws IgniteCheckedException If cancelled.
*/
- private void checkCancelled() throws SchemaIndexOperationCancellationException {
- if (nonNull(cancel) && cancel.isCancelled())
- throw new SchemaIndexOperationCancellationException("Index creation was cancelled.");
+ private void checkCancelled() throws IgniteCheckedException {
+ Throwable e = cancelTok.cancelException();
+
+ if (e instanceof SchemaIndexOperationCancellationException)
+ throw (SchemaIndexOperationCancellationException)e;
+ else if (e != null)
+ throw new IgniteCheckedException(e);
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexCacheVisitorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexCacheVisitorImpl.java
index b69b8bb..acb17b6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexCacheVisitorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexCacheVisitorImpl.java
@@ -37,7 +37,6 @@ import org.apache.ignite.internal.util.typedef.internal.SB;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.util.worker.GridWorker;
import org.apache.ignite.internal.util.worker.GridWorkerFuture;
-import org.jetbrains.annotations.Nullable;
import static java.util.Objects.isNull;
import static java.util.Objects.nonNull;
@@ -55,7 +54,7 @@ public class SchemaIndexCacheVisitorImpl implements SchemaIndexCacheVisitor {
private final GridCacheContext cctx;
/** Cancellation token. */
- @Nullable private final SchemaIndexOperationCancellationToken cancel;
+ private final IndexRebuildCancelToken cancelTok;
/** Future for create/rebuild index. */
protected final GridFutureAdapter<Void> buildIdxFut;
@@ -67,12 +66,12 @@ public class SchemaIndexCacheVisitorImpl implements SchemaIndexCacheVisitor {
* Constructor.
*
* @param cctx Cache context.
- * @param cancel Cancellation token.
+ * @param cancelTok Cancellation token.
* @param buildIdxFut Future for create/rebuild index.
*/
public SchemaIndexCacheVisitorImpl(
- GridCacheContext cctx,
- @Nullable SchemaIndexOperationCancellationToken cancel,
+ GridCacheContext<?, ?> cctx,
+ IndexRebuildCancelToken cancelTok,
GridFutureAdapter<Void> buildIdxFut
) {
assert nonNull(cctx);
@@ -84,7 +83,7 @@ public class SchemaIndexCacheVisitorImpl implements SchemaIndexCacheVisitor {
this.cctx = cctx;
this.buildIdxFut = buildIdxFut;
- this.cancel = cancel;
+ this.cancelTok = cancelTok;
log = cctx.kernalContext().log(getClass());
}
@@ -118,7 +117,7 @@ public class SchemaIndexCacheVisitorImpl implements SchemaIndexCacheVisitor {
GridWorkerFuture<SchemaIndexCacheStat> workerFut = new GridWorkerFuture<>();
GridWorker worker =
- new SchemaIndexCachePartitionWorker(cctx, locPart, stop, cancel, clo, workerFut, partsCnt);
+ new SchemaIndexCachePartitionWorker(cctx, locPart, stop, cancelTok, clo, workerFut, partsCnt);
workerFut.setWorker(worker);
buildIdxCompoundFut.add(workerFut);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexOperationCancellationException.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexOperationCancellationException.java
index 360fc01..3f28907 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexOperationCancellationException.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaIndexOperationCancellationException.java
@@ -19,7 +19,7 @@ package org.apache.ignite.internal.processors.query.schema;
import org.apache.ignite.IgniteCheckedException;
/**
- * Exception occurred when canceling index rebuild via {@link SchemaIndexOperationCancellationToken}.
+ * Exception occurred when canceling index rebuild via {@link IndexRebuildCancelToken}.
*/
public class SchemaIndexOperationCancellationException extends IgniteCheckedException {
/** Serial version uid. */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaOperationWorker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaOperationWorker.java
index 5f58c6e..6df7dd7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaOperationWorker.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/schema/SchemaOperationWorker.java
@@ -67,7 +67,7 @@ public class SchemaOperationWorker extends GridWorker {
private final AtomicBoolean startGuard = new AtomicBoolean();
/** Cancellation token. */
- private final SchemaIndexOperationCancellationToken cancelToken = new SchemaIndexOperationCancellationToken();
+ private final IndexRebuildCancelToken cancelTok = new IndexRebuildCancelToken();
/** Workers registry. */
private final WorkersRegistry workersRegistry;
@@ -111,7 +111,7 @@ public class SchemaOperationWorker extends GridWorker {
@Override protected void body() throws InterruptedException, IgniteInterruptedCheckedException {
try {
// Execute.
- qryProc.processSchemaOperationLocal(op, type, depId, cancelToken);
+ qryProc.processSchemaOperationLocal(op, type, depId, cancelTok);
fut.onDone();
}
@@ -184,7 +184,7 @@ public class SchemaOperationWorker extends GridWorker {
* Cancel operation.
*/
@Override public void cancel() {
- if (cancelToken.cancel()) {
+ if (cancelTok.cancel(new SchemaIndexOperationCancellationException("Index creation was cancelled."))) {
try {
fut.get(workersRegistry.getSystemWorkerBlockedTimeout());
}
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/AbstractIndexingCommonTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/AbstractIndexingCommonTest.java
index 00057cb3..b60acfd7 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/AbstractIndexingCommonTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/AbstractIndexingCommonTest.java
@@ -34,9 +34,9 @@ import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStor
import org.apache.ignite.internal.processors.query.h2.ConnectionManager;
import org.apache.ignite.internal.processors.query.h2.H2PooledConnection;
import org.apache.ignite.internal.processors.query.h2.IgniteH2Indexing;
+import org.apache.ignite.internal.processors.query.schema.IndexRebuildCancelToken;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorClosure;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorImpl;
-import org.apache.ignite.internal.processors.query.schema.SchemaIndexOperationCancellationToken;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.typedef.G;
@@ -157,7 +157,7 @@ public class AbstractIndexingCommonTest extends GridCommonAbstractTest {
GridCacheContext cctx,
GridFutureAdapter<Void> rebuildIdxFut,
SchemaIndexCacheVisitorClosure clo,
- SchemaIndexOperationCancellationToken cancel
+ IndexRebuildCancelToken cancel
) {
CountDownLatch startThread = new CountDownLatch(1);
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/DynamicEnableIndexingConcurrentSelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/DynamicEnableIndexingConcurrentSelfTest.java
index 94c2f0f..f4e3a1d 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/DynamicEnableIndexingConcurrentSelfTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/DynamicEnableIndexingConcurrentSelfTest.java
@@ -52,8 +52,8 @@ import org.apache.ignite.internal.managers.discovery.CustomEventListener;
import org.apache.ignite.internal.managers.indexing.IndexesRebuildTask;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.query.schema.IndexRebuildCancelToken;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorClosure;
-import org.apache.ignite.internal.processors.query.schema.SchemaIndexOperationCancellationToken;
import org.apache.ignite.internal.processors.query.schema.SchemaOperationException;
import org.apache.ignite.internal.processors.query.schema.message.SchemaFinishDiscoveryMessage;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
@@ -632,7 +632,7 @@ public class DynamicEnableIndexingConcurrentSelfTest extends DynamicEnableIndexi
/** {@inheritDoc} */
@Override public void startRebuild(
GridCacheContext cctx, GridFutureAdapter<Void> fut, SchemaIndexCacheVisitorClosure clo,
- SchemaIndexOperationCancellationToken cancel) {
+ IndexRebuildCancelToken cancel) {
awaitIndexing(cctx.localNodeId());
super.startRebuild(cctx, fut, clo, cancel);
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/IndexesRebuildTaskEx.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/IndexesRebuildTaskEx.java
index d182b46..a3fcb52 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/IndexesRebuildTaskEx.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/IndexesRebuildTaskEx.java
@@ -27,8 +27,8 @@ import org.apache.ignite.internal.cache.query.index.IndexProcessor;
import org.apache.ignite.internal.managers.indexing.IndexesRebuildTask;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
+import org.apache.ignite.internal.processors.query.schema.IndexRebuildCancelToken;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorClosure;
-import org.apache.ignite.internal.processors.query.schema.SchemaIndexOperationCancellationToken;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.lang.IgniteThrowableConsumer;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
@@ -60,7 +60,7 @@ class IndexesRebuildTaskEx extends IndexesRebuildTask {
GridCacheContext cctx,
GridFutureAdapter<Void> rebuildIdxFut,
SchemaIndexCacheVisitorClosure clo,
- SchemaIndexOperationCancellationToken cancel
+ IndexRebuildCancelToken cancel
) {
super.startRebuild(cctx, rebuildIdxFut, new SchemaIndexCacheVisitorClosure() {
/** {@inheritDoc} */
@@ -76,10 +76,14 @@ class IndexesRebuildTaskEx extends IndexesRebuildTask {
}
/** {@inheritDoc} */
- @Override @Nullable public IgniteInternalFuture<?> rebuild(GridCacheContext cctx, boolean force) {
+ @Override @Nullable public IgniteInternalFuture<?> rebuild(
+ GridCacheContext cctx,
+ boolean force,
+ IndexRebuildCancelToken cancelTok
+ ) {
cacheRebuildRunner.getOrDefault(nodeName(cctx), emptyMap()).getOrDefault(cctx.name(), () -> { }).run();
- return super.rebuild(cctx, force);
+ return super.rebuild(cctx, force, cancelTok);
}
/**
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/StopRebuildIndexTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/StopRebuildIndexTest.java
index 48e0e28..b34700c 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/StopRebuildIndexTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/index/StopRebuildIndexTest.java
@@ -163,14 +163,14 @@ public class StopRebuildIndexTest extends AbstractRebuildIndexTest {
assertFalse(rebFut0.isDone());
assertFalse(rebFut1.isDone());
- assertFalse(rebFut1.cancelToken().isCancelled());
+ assertNull(rebFut1.cancelToken().cancelException());
stopRebuildIdxConsumer.finishBuildIdxFut.onDone();
rebFut0.get(getTestTimeout());
rebFut1.get(getTestTimeout());
- assertFalse(rebFut1.cancelToken().isCancelled());
+ assertNull(rebFut1.cancelToken().cancelException());
assertNull(indexRebuildFuture(n, cacheCtx.cacheId()));
assertNull(internalIndexRebuildFuture(n, cacheCtx.cacheId()));
@@ -212,14 +212,14 @@ public class StopRebuildIndexTest extends AbstractRebuildIndexTest {
assertFalse(fut0.isDone());
assertFalse(fut1.isDone());
- assertFalse(fut1.cancelToken().isCancelled());
+ assertNull(fut1.cancelToken().cancelException());
assertTrue(waitForCondition(() -> metrics0.getIndexRebuildKeysProcessed() >= keys / 100, getTestTimeout()));
assertTrue(metrics0.isIndexRebuildInProgress());
assertFalse(fut0.isDone());
assertFalse(fut1.isDone());
- assertFalse(fut1.cancelToken().isCancelled());
+ assertNull(fut1.cancelToken().cancelException());
stopRebuildIndexes.accept(n);
@@ -230,13 +230,13 @@ public class StopRebuildIndexTest extends AbstractRebuildIndexTest {
assertThrows(log, () -> fut0.get(getTestTimeout()), SchemaIndexOperationCancellationException.class, null);
assertThrows(log, () -> fut1.get(getTestTimeout()), SchemaIndexOperationCancellationException.class, null);
- assertTrue(fut1.cancelToken().isCancelled());
+ assertNotNull(fut1.cancelToken().cancelException());
}
else {
fut0.get(getTestTimeout());
fut1.get(getTestTimeout());
- assertFalse(fut1.cancelToken().isCancelled());
+ assertNull(fut1.cancelToken().cancelException());
}
assertNull(internalIndexRebuildFuture(n, cacheCtx.cacheId()));
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsIndexingDefragmentationTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsIndexingDefragmentationTest.java
index c4cda6c..c25537d 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsIndexingDefragmentationTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/IgnitePdsIndexingDefragmentationTest.java
@@ -38,6 +38,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.IgniteCacheUpdateSqlQuerySelfTest;
import org.apache.ignite.internal.processors.cache.persistence.defragmentation.DefragmentationFileUtils;
import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager;
+import org.apache.ignite.internal.processors.query.schema.IndexRebuildCancelToken;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.visor.verify.ValidateIndexesClosure;
@@ -311,8 +312,12 @@ public class IgnitePdsIndexingDefragmentationTest extends IgnitePdsDefragmentati
private boolean rebuiltIndexes;
/** {@inheritDoc} */
- @Override @Nullable public IgniteInternalFuture<?> rebuild(GridCacheContext cctx, boolean force) {
- IgniteInternalFuture<?> future = super.rebuild(cctx, force);
+ @Override @Nullable public IgniteInternalFuture<?> rebuild(
+ GridCacheContext cctx,
+ boolean force,
+ IndexRebuildCancelToken cancelTok
+ ) {
+ IgniteInternalFuture<?> future = super.rebuild(cctx, force, cancelTok);
rebuiltIndexes = future != null;
return future;
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/GridIndexRebuildSelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/GridIndexRebuildSelfTest.java
index f83c7d1..73df179 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/GridIndexRebuildSelfTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/h2/GridIndexRebuildSelfTest.java
@@ -38,8 +38,8 @@ import org.apache.ignite.internal.processors.cache.index.DynamicIndexAbstractSel
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager;
import org.apache.ignite.internal.processors.query.h2.opt.GridH2Table;
+import org.apache.ignite.internal.processors.query.schema.IndexRebuildCancelToken;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorClosure;
-import org.apache.ignite.internal.processors.query.schema.SchemaIndexOperationCancellationToken;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.lang.GridCursor;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -398,7 +398,7 @@ public class GridIndexRebuildSelfTest extends DynamicIndexAbstractSelfTest {
GridCacheContext cctx,
GridFutureAdapter<Void> rebuildIdxFut,
SchemaIndexCacheVisitorClosure clo,
- SchemaIndexOperationCancellationToken cancel
+ IndexRebuildCancelToken cancel
) {
if (!firstRbld) {
try {