You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ap...@apache.org on 2024/02/19 11:34:27 UTC
(ignite-3) branch main updated: IGNITE-21117 Wait for completion of transactions before index drop (#3216)
This is an automated email from the ASF dual-hosted git repository.
apolovtsev pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 5068c3acf1 IGNITE-21117 Wait for completion of transactions before index drop (#3216)
5068c3acf1 is described below
commit 5068c3acf1e821dbf4ba9b3f6c8ada2b4649c6a6
Author: Alexander Polovtcev <al...@gmail.com>
AuthorDate: Mon Feb 19 13:34:22 2024 +0200
IGNITE-21117 Wait for completion of transactions before index drop (#3216)
---
.../internal/catalog/commands/CatalogUtils.java | 31 +++
.../descriptors/CatalogHashIndexDescriptor.java | 27 +--
.../descriptors/CatalogIndexDescriptor.java | 19 +-
.../descriptors/CatalogSortedIndexDescriptor.java | 27 +--
.../storage/AbstractChangeIndexStatusEntry.java | 43 +++--
.../catalog/CatalogManagerRecoveryTest.java | 2 +-
.../internal/catalog/CatalogManagerSelfTest.java | 10 +-
.../catalog/commands/CatalogUtilsTest.java | 52 ++++-
.../ignite/internal/event/EventListener.java | 23 +++
.../distributionzones/DistributionZoneManager.java | 3 +-
modules/index/build.gradle | 1 +
.../ignite/internal/index/ItBuildIndexTest.java | 213 ++++++++++++++++++++-
...StarterTask.java => ChangeIndexStatusTask.java} | 147 ++++++++------
...r.java => ChangeIndexStatusTaskController.java} | 127 ++++++------
...terTaskId.java => ChangeIndexStatusTaskId.java} | 33 ++--
.../index/ChangeIndexStatusTaskScheduler.java | 213 +++++++++++++++++++++
.../internal/index/IndexBuildController.java | 2 +-
.../internal/index/IndexBuildingManager.java | 21 +-
.../internal/index/IndexBuildingStarter.java | 163 ----------------
.../apache/ignite/internal/index/IndexChooser.java | 35 +---
.../apache/ignite/internal/index/IndexManager.java | 46 +----
...va => ChangeIndexStatusTaskControllerTest.java} | 66 +++++--
...askTest.java => ChangeIndexStatusTaskTest.java} | 32 ++--
.../index/IndexAvailabilityControllerTest.java | 2 +-
.../internal/index/IndexBuildControllerTest.java | 2 +-
.../replicator/PartitionReplicaListener.java | 4 +-
.../replication/PartitionReplicaListenerTest.java | 2 +-
27 files changed, 865 insertions(+), 481 deletions(-)
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java
index 502b922760..016843cd8c 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java
@@ -343,6 +343,37 @@ public class CatalogUtils {
));
}
+ /**
+ * Replaces the index descriptor that has the same ID as the {@code newIndexDescriptor} in the given {@code schema}.
+ *
+ * @param schema Schema, which index descriptor needs to be replaced.
+ * @param newIndexDescriptor Index descriptor which will replace the descriptor with the same ID in the schema.
+ * @return New schema descriptor with a replaced index descriptor.
+ * @throws CatalogValidationException If the index descriptor with the same ID is not present in the schema.
+ */
+ public static CatalogSchemaDescriptor replaceIndex(CatalogSchemaDescriptor schema, CatalogIndexDescriptor newIndexDescriptor) {
+ CatalogIndexDescriptor[] indexDescriptors = schema.indexes().clone();
+
+ for (int i = 0; i < indexDescriptors.length; i++) {
+ if (indexDescriptors[i].id() == newIndexDescriptor.id()) {
+ indexDescriptors[i] = newIndexDescriptor;
+
+ return new CatalogSchemaDescriptor(
+ schema.id(),
+ schema.name(),
+ schema.tables(),
+ indexDescriptors,
+ schema.systemViews(),
+ newIndexDescriptor.updateToken()
+ );
+ }
+ }
+
+ throw new CatalogValidationException(String.format(
+ "Index with ID %d has not been found in schema with ID %d", newIndexDescriptor.id(), schema.id()
+ ));
+ }
+
/**
* Return default length according to supplied type.
*
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogHashIndexDescriptor.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogHashIndexDescriptor.java
index 3c35c024d4..35e86abd03 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogHashIndexDescriptor.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogHashIndexDescriptor.java
@@ -43,12 +43,13 @@ public class CatalogHashIndexDescriptor extends CatalogIndexDescriptor {
* @param name Name of the index.
* @param tableId Id of the table index belongs to.
* @param unique Unique flag.
- * @param creationCatalogVersion Catalog version in which the index was created.
+ * @param txWaitCatalogVersion Catalog version used in special index status updates to wait for RW transactions, started before
+ * this version, to finish.
* @param columns A list of indexed columns. Must not contains duplicates.
* @throws IllegalArgumentException If columns list contains duplicates.
*/
- public CatalogHashIndexDescriptor(int id, String name, int tableId, boolean unique, int creationCatalogVersion, List<String> columns) {
- this(id, name, tableId, unique, CatalogIndexStatus.REGISTERED, creationCatalogVersion, columns, INITIAL_CAUSALITY_TOKEN);
+ public CatalogHashIndexDescriptor(int id, String name, int tableId, boolean unique, int txWaitCatalogVersion, List<String> columns) {
+ this(id, name, tableId, unique, CatalogIndexStatus.REGISTERED, txWaitCatalogVersion, columns, INITIAL_CAUSALITY_TOKEN);
}
/**
@@ -59,7 +60,8 @@ public class CatalogHashIndexDescriptor extends CatalogIndexDescriptor {
* @param tableId Id of the table index belongs to.
* @param unique Unique flag.
* @param status Index status.
- * @param creationCatalogVersion Catalog version in which the index was created.
+ * @param txWaitCatalogVersion Catalog version used in special index status updates to wait for RW transactions, started before
+ * this version, to finish.
* @param columns A list of indexed columns. Must not contains duplicates.
* @throws IllegalArgumentException If columns list contains duplicates.
*/
@@ -69,10 +71,10 @@ public class CatalogHashIndexDescriptor extends CatalogIndexDescriptor {
int tableId,
boolean unique,
CatalogIndexStatus status,
- int creationCatalogVersion,
+ int txWaitCatalogVersion,
List<String> columns
) {
- this(id, name, tableId, unique, status, creationCatalogVersion, columns, INITIAL_CAUSALITY_TOKEN);
+ this(id, name, tableId, unique, status, txWaitCatalogVersion, columns, INITIAL_CAUSALITY_TOKEN);
}
/**
@@ -83,7 +85,8 @@ public class CatalogHashIndexDescriptor extends CatalogIndexDescriptor {
* @param tableId Id of the table index belongs to.
* @param unique Unique flag.
* @param status Index status.
- * @param creationCatalogVersion Catalog version in which the index was created.
+ * @param txWaitCatalogVersion Catalog version used in special index status updates to wait for RW transactions, started before
+ * this version, to finish.
* @param columns A list of indexed columns. Must not contains duplicates.
* @param causalityToken Token of the update of the descriptor.
* @throws IllegalArgumentException If columns list contains duplicates.
@@ -94,11 +97,11 @@ public class CatalogHashIndexDescriptor extends CatalogIndexDescriptor {
int tableId,
boolean unique,
CatalogIndexStatus status,
- int creationCatalogVersion,
+ int txWaitCatalogVersion,
List<String> columns,
long causalityToken
) {
- super(CatalogIndexDescriptorType.HASH, id, name, tableId, unique, status, creationCatalogVersion, causalityToken);
+ super(CatalogIndexDescriptorType.HASH, id, name, tableId, unique, status, txWaitCatalogVersion, causalityToken);
this.columns = List.copyOf(Objects.requireNonNull(columns, "columns"));
}
@@ -122,10 +125,10 @@ public class CatalogHashIndexDescriptor extends CatalogIndexDescriptor {
int tableId = input.readInt();
boolean unique = input.readBoolean();
CatalogIndexStatus status = CatalogIndexStatus.forId(input.readByte());
- int creationCatalogVersion = input.readInt();
+ int txWaitCatalogVersion = input.readInt();
List<String> columns = readStringCollection(input, ArrayList::new);
- return new CatalogHashIndexDescriptor(id, name, tableId, unique, status, creationCatalogVersion, columns, updateToken);
+ return new CatalogHashIndexDescriptor(id, name, tableId, unique, status, txWaitCatalogVersion, columns, updateToken);
}
@Override
@@ -136,7 +139,7 @@ public class CatalogHashIndexDescriptor extends CatalogIndexDescriptor {
output.writeInt(descriptor.tableId());
output.writeBoolean(descriptor.unique());
output.writeByte(descriptor.status().id());
- output.writeInt(descriptor.creationCatalogVersion());
+ output.writeInt(descriptor.txWaitCatalogVersion());
writeStringCollection(descriptor.columns(), output);
}
}
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogIndexDescriptor.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogIndexDescriptor.java
index 35fb5b46c5..93a1affae6 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogIndexDescriptor.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogIndexDescriptor.java
@@ -31,20 +31,22 @@ public abstract class CatalogIndexDescriptor extends CatalogObjectDescriptor {
/** Index status. */
private final CatalogIndexStatus status;
- /** Catalog version in which the index was created. */
- private final int creationCatalogVersion;
+ /**
+ * Catalog version used in special index status updates to wait for RW transactions, started before this version, to finish.
+ */
+ private final int txWaitCatalogVersion;
/** Index descriptor type. */
private final CatalogIndexDescriptorType indexType;
CatalogIndexDescriptor(CatalogIndexDescriptorType indexType, int id, String name, int tableId, boolean unique,
- CatalogIndexStatus status, int creationCatalogVersion, long causalityToken) {
+ CatalogIndexStatus status, int txWaitCatalogVersion, long causalityToken) {
super(id, Type.INDEX, name, causalityToken);
this.indexType = indexType;
this.tableId = tableId;
this.unique = unique;
this.status = Objects.requireNonNull(status, "status");
- this.creationCatalogVersion = creationCatalogVersion;
+ this.txWaitCatalogVersion = txWaitCatalogVersion;
}
/** Gets table ID. */
@@ -62,9 +64,12 @@ public abstract class CatalogIndexDescriptor extends CatalogObjectDescriptor {
return status;
}
- /** Returns catalog version in which the index was created. */
- public int creationCatalogVersion() {
- return creationCatalogVersion;
+ /**
+ * Returns the Catalog version used in special index status updates to wait for RW transactions, started before
+ * this version, to finish.
+ */
+ public int txWaitCatalogVersion() {
+ return txWaitCatalogVersion;
}
/** Returns catalog index descriptor type. */
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogSortedIndexDescriptor.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogSortedIndexDescriptor.java
index 7bee87d7fa..6697c16259 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogSortedIndexDescriptor.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/descriptors/CatalogSortedIndexDescriptor.java
@@ -44,7 +44,8 @@ public class CatalogSortedIndexDescriptor extends CatalogIndexDescriptor {
* @param name Name of the index.
* @param tableId Id of the table index belongs to.
* @param unique Unique flag.
- * @param creationCatalogVersion Catalog version in which the index was created.
+ * @param txWaitCatalogVersion Catalog version used in special index status updates to wait for RW transactions, started before
+ * this version, to finish.
* @param columns A list of columns descriptors.
* @throws IllegalArgumentException If columns list contains duplicates or columns size doesn't match the collations size.
*/
@@ -53,10 +54,10 @@ public class CatalogSortedIndexDescriptor extends CatalogIndexDescriptor {
String name,
int tableId,
boolean unique,
- int creationCatalogVersion,
+ int txWaitCatalogVersion,
List<CatalogIndexColumnDescriptor> columns
) {
- this(id, name, tableId, unique, REGISTERED, creationCatalogVersion, columns);
+ this(id, name, tableId, unique, REGISTERED, txWaitCatalogVersion, columns);
}
/**
@@ -67,7 +68,8 @@ public class CatalogSortedIndexDescriptor extends CatalogIndexDescriptor {
* @param tableId Id of the table index belongs to.
* @param unique Unique flag.
* @param status Index status.
- * @param creationCatalogVersion Catalog version in which the index was created.
+ * @param txWaitCatalogVersion Catalog version used in special index status updates to wait for RW transactions, started before
+ * this version, to finish.
* @param columns A list of columns descriptors.
* @throws IllegalArgumentException If columns list contains duplicates or columns size doesn't match the collations size.
*/
@@ -77,10 +79,10 @@ public class CatalogSortedIndexDescriptor extends CatalogIndexDescriptor {
int tableId,
boolean unique,
CatalogIndexStatus status,
- int creationCatalogVersion,
+ int txWaitCatalogVersion,
List<CatalogIndexColumnDescriptor> columns
) {
- this(id, name, tableId, unique, status, creationCatalogVersion, columns, INITIAL_CAUSALITY_TOKEN);
+ this(id, name, tableId, unique, status, txWaitCatalogVersion, columns, INITIAL_CAUSALITY_TOKEN);
}
/**
@@ -91,7 +93,8 @@ public class CatalogSortedIndexDescriptor extends CatalogIndexDescriptor {
* @param tableId Id of the table index belongs to.
* @param unique Unique flag.
* @param status Index status.
- * @param creationCatalogVersion Catalog version in which the index was created.
+ * @param txWaitCatalogVersion Catalog version used in special index status updates to wait for RW transactions, started before
+ * this version, to finish.
* @param columns A list of columns descriptors.
* @param causalityToken Token of the update of the descriptor.
* @throws IllegalArgumentException If columns list contains duplicates or columns size doesn't match the collations size.
@@ -102,11 +105,11 @@ public class CatalogSortedIndexDescriptor extends CatalogIndexDescriptor {
int tableId,
boolean unique,
CatalogIndexStatus status,
- int creationCatalogVersion,
+ int txWaitCatalogVersion,
List<CatalogIndexColumnDescriptor> columns,
long causalityToken
) {
- super(CatalogIndexDescriptorType.SORTED, id, name, tableId, unique, status, creationCatalogVersion, causalityToken);
+ super(CatalogIndexDescriptorType.SORTED, id, name, tableId, unique, status, txWaitCatalogVersion, causalityToken);
this.columns = Objects.requireNonNull(columns, "columns");
}
@@ -130,10 +133,10 @@ public class CatalogSortedIndexDescriptor extends CatalogIndexDescriptor {
int tableId = input.readInt();
boolean unique = input.readBoolean();
CatalogIndexStatus status = CatalogIndexStatus.forId(input.readByte());
- int creationCatalogVersion = input.readInt();
+ int txWaitCatalogVersion = input.readInt();
List<CatalogIndexColumnDescriptor> columns = readList(CatalogIndexColumnDescriptor.SERIALIZER, input);
- return new CatalogSortedIndexDescriptor(id, name, tableId, unique, status, creationCatalogVersion, columns, updateToken);
+ return new CatalogSortedIndexDescriptor(id, name, tableId, unique, status, txWaitCatalogVersion, columns, updateToken);
}
@Override
@@ -144,7 +147,7 @@ public class CatalogSortedIndexDescriptor extends CatalogIndexDescriptor {
output.writeInt(descriptor.tableId());
output.writeBoolean(descriptor.unique());
output.writeByte(descriptor.status().id());
- output.writeInt(descriptor.creationCatalogVersion());
+ output.writeInt(descriptor.txWaitCatalogVersion());
writeList(descriptor.columns(), CatalogIndexColumnDescriptor.SERIALIZER, output);
}
}
diff --git a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/AbstractChangeIndexStatusEntry.java b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/AbstractChangeIndexStatusEntry.java
index 1199372aa6..484eeb4730 100644
--- a/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/AbstractChangeIndexStatusEntry.java
+++ b/modules/catalog/src/main/java/org/apache/ignite/internal/catalog/storage/AbstractChangeIndexStatusEntry.java
@@ -17,12 +17,13 @@
package org.apache.ignite.internal.catalog.storage;
+import static org.apache.ignite.internal.catalog.commands.CatalogUtils.indexOrThrow;
+import static org.apache.ignite.internal.catalog.commands.CatalogUtils.replaceIndex;
+import static org.apache.ignite.internal.catalog.commands.CatalogUtils.replaceSchema;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
-import java.util.Arrays;
import org.apache.ignite.internal.catalog.Catalog;
import org.apache.ignite.internal.catalog.CatalogValidationException;
-import org.apache.ignite.internal.catalog.commands.CatalogUtils;
import org.apache.ignite.internal.catalog.descriptors.CatalogHashIndexDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
@@ -46,21 +47,14 @@ abstract class AbstractChangeIndexStatusEntry implements UpdateEntry {
public final Catalog applyUpdate(Catalog catalog, long causalityToken) {
CatalogSchemaDescriptor schema = schemaByIndexId(catalog, indexId);
+ CatalogIndexDescriptor newIndexDescriptor = updateIndexStatus(catalog, causalityToken, newStatus);
+
return new Catalog(
catalog.version(),
catalog.time(),
catalog.objectIdGenState(),
catalog.zones(),
- CatalogUtils.replaceSchema(new CatalogSchemaDescriptor(
- schema.id(),
- schema.name(),
- schema.tables(),
- Arrays.stream(schema.indexes())
- .map(source -> source.id() == indexId ? updateIndexStatus(source, causalityToken, newStatus) : source)
- .toArray(CatalogIndexDescriptor[]::new),
- schema.systemViews(),
- causalityToken
- ), catalog.schemas())
+ replaceSchema(replaceIndex(schema, newIndexDescriptor), catalog.schemas())
);
}
@@ -80,17 +74,22 @@ abstract class AbstractChangeIndexStatusEntry implements UpdateEntry {
return schema;
}
- private static CatalogIndexDescriptor updateIndexStatus(
- CatalogIndexDescriptor source,
+ private CatalogIndexDescriptor updateIndexStatus(
+ Catalog catalog,
long causalityToken,
CatalogIndexStatus newStatus
) {
+ CatalogIndexDescriptor source = indexOrThrow(catalog, indexId);
+
+ // We only care about the transitions to REGISTERED and STOPPING. REGISTERED status has already been handled on index creation.
+ int txWaitCatalogVersion = newStatus == CatalogIndexStatus.STOPPING ? catalog.version() + 1 : source.txWaitCatalogVersion();
+
CatalogIndexDescriptor updateIndexDescriptor;
if (source instanceof CatalogHashIndexDescriptor) {
- updateIndexDescriptor = updateHashIndexStatus((CatalogHashIndexDescriptor) source, newStatus);
+ updateIndexDescriptor = updateHashIndexStatus((CatalogHashIndexDescriptor) source, newStatus, txWaitCatalogVersion);
} else if (source instanceof CatalogSortedIndexDescriptor) {
- updateIndexDescriptor = updateSortedIndexStatus((CatalogSortedIndexDescriptor) source, newStatus);
+ updateIndexDescriptor = updateSortedIndexStatus((CatalogSortedIndexDescriptor) source, newStatus, txWaitCatalogVersion);
} else {
throw new CatalogValidationException(format("Unsupported index type '{}' {}", source.id(), source));
}
@@ -100,26 +99,30 @@ abstract class AbstractChangeIndexStatusEntry implements UpdateEntry {
return updateIndexDescriptor;
}
- private static CatalogIndexDescriptor updateHashIndexStatus(CatalogHashIndexDescriptor index, CatalogIndexStatus newStatus) {
+ private static CatalogIndexDescriptor updateHashIndexStatus(
+ CatalogHashIndexDescriptor index, CatalogIndexStatus newStatus, int txWaitCatalogVersion
+ ) {
return new CatalogHashIndexDescriptor(
index.id(),
index.name(),
index.tableId(),
index.unique(),
newStatus,
- index.creationCatalogVersion(),
+ txWaitCatalogVersion,
index.columns()
);
}
- private static CatalogIndexDescriptor updateSortedIndexStatus(CatalogSortedIndexDescriptor index, CatalogIndexStatus newStatus) {
+ private static CatalogIndexDescriptor updateSortedIndexStatus(
+ CatalogSortedIndexDescriptor index, CatalogIndexStatus newStatus, int txWaitCatalogVersion
+ ) {
return new CatalogSortedIndexDescriptor(
index.id(),
index.name(),
index.tableId(),
index.unique(),
newStatus,
- index.creationCatalogVersion(),
+ txWaitCatalogVersion,
index.columns()
);
}
diff --git a/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerRecoveryTest.java b/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerRecoveryTest.java
index e1a53b1f27..dda29e1b80 100644
--- a/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerRecoveryTest.java
+++ b/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerRecoveryTest.java
@@ -165,7 +165,7 @@ public class CatalogManagerRecoveryTest extends BaseIgniteAbstractTest {
createAndStartComponents();
- assertEquals(expCreationCatalogVersion, catalogManager.index(INDEX_NAME, clock.nowLong()).creationCatalogVersion());
+ assertEquals(expCreationCatalogVersion, catalogManager.index(INDEX_NAME, clock.nowLong()).txWaitCatalogVersion());
}
private void createAndStartComponents() {
diff --git a/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerSelfTest.java b/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerSelfTest.java
index 7600345f5c..3f296fe792 100644
--- a/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerSelfTest.java
+++ b/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/CatalogManagerSelfTest.java
@@ -287,7 +287,7 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
assertEquals(table.primaryKeyColumns(), pkIndex.columns());
assertTrue(pkIndex.unique());
assertEquals(AVAILABLE, pkIndex.status());
- assertEquals(manager.latestCatalogVersion(), pkIndex.creationCatalogVersion());
+ assertEquals(manager.latestCatalogVersion(), pkIndex.txWaitCatalogVersion());
CatalogTableColumnDescriptor desc = table.columnDescriptor("key1");
assertNotNull(desc);
@@ -1012,7 +1012,7 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
assertEquals(List.of("VAL", "ID"), index.columns());
assertFalse(index.unique());
assertEquals(REGISTERED, index.status());
- assertEquals(manager.latestCatalogVersion(), index.creationCatalogVersion());
+ assertEquals(manager.latestCatalogVersion(), index.txWaitCatalogVersion());
}
@Test
@@ -1054,7 +1054,7 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
assertEquals(ASC_NULLS_LAST, index.columns().get(1).collation());
assertTrue(index.unique());
assertEquals(REGISTERED, index.status());
- assertEquals(manager.latestCatalogVersion(), index.creationCatalogVersion());
+ assertEquals(manager.latestCatalogVersion(), index.txWaitCatalogVersion());
}
@Test
@@ -2416,7 +2416,7 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
assertThat(latestCatalog.version(), greaterThan(expCreationVersion));
- assertEquals(expCreationVersion, latestCatalog.index(indexId).creationCatalogVersion());
+ assertEquals(expCreationVersion, latestCatalog.index(indexId).txWaitCatalogVersion());
}
@ParameterizedTest(name = "hashIndex={0}, updateIndex={1}")
@@ -2444,7 +2444,7 @@ public class CatalogManagerSelfTest extends BaseCatalogManagerTest {
assertThat(latestCatalog.version(), greaterThan(expCreationVersion));
- assertEquals(expCreationVersion, latestCatalog.index(indexId).creationCatalogVersion());
+ assertEquals(expCreationVersion, latestCatalog.index(indexId).txWaitCatalogVersion());
}
private static Stream<Arguments> argumentsForCheckIndexCreationCatalogVersion() {
diff --git a/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/commands/CatalogUtilsTest.java b/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/commands/CatalogUtilsTest.java
index 28232ec47e..28550dc094 100644
--- a/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/commands/CatalogUtilsTest.java
+++ b/modules/catalog/src/test/java/org/apache/ignite/internal/catalog/commands/CatalogUtilsTest.java
@@ -25,6 +25,7 @@ import static org.apache.ignite.internal.catalog.CatalogTestUtils.index;
import static org.apache.ignite.internal.catalog.commands.CatalogUtils.clusterWideEnsuredActivationTimestamp;
import static org.apache.ignite.internal.catalog.commands.CatalogUtils.collectIndexes;
import static org.apache.ignite.internal.catalog.commands.CatalogUtils.pkIndexName;
+import static org.apache.ignite.internal.catalog.commands.CatalogUtils.replaceIndex;
import static org.apache.ignite.internal.catalog.commands.CatalogUtils.replaceTable;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.sql.ColumnType.INT32;
@@ -47,6 +48,7 @@ import org.apache.ignite.internal.catalog.Catalog;
import org.apache.ignite.internal.catalog.CatalogCommand;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.CatalogValidationException;
+import org.apache.ignite.internal.catalog.descriptors.CatalogHashIndexDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogSchemaDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
@@ -336,7 +338,55 @@ public class CatalogUtilsTest extends BaseIgniteAbstractTest {
Exception e = assertThrows(CatalogValidationException.class, () -> replaceTable(schema, table));
- assertThat(e.getMessage(), is(String.format("Table with ID %d has not been found in schema with ID %d", Integer.MAX_VALUE, 0)));
+ assertThat(e.getMessage(), is(String.format("Table with ID %d has not been found in schema with ID %d", table.id(), 0)));
+ }
+
+ @Test
+ void testReplaceIndex() {
+ String tableName = "table";
+
+ createTable(tableName);
+ createIndex(tableName, "foo");
+ createIndex(tableName, "bar");
+
+ CatalogSchemaDescriptor schema = catalogManager.activeSchema(DEFAULT_SCHEMA_NAME, clock.nowLong());
+
+ assertThat(schema, is(notNullValue()));
+
+ var fooIndex = (CatalogHashIndexDescriptor) catalogManager.index("foo", clock.nowLong());
+
+ assertThat(fooIndex, is(notNullValue()));
+
+ CatalogIndexDescriptor bazIndex = new CatalogHashIndexDescriptor(
+ fooIndex.id(),
+ "baz",
+ fooIndex.tableId(),
+ fooIndex.unique(),
+ fooIndex.status(),
+ fooIndex.txWaitCatalogVersion(),
+ fooIndex.columns()
+ );
+
+ CatalogSchemaDescriptor updatedSchema = replaceIndex(schema, bazIndex);
+
+ List<String> indexNames = Arrays.stream(updatedSchema.indexes()).map(CatalogIndexDescriptor::name).collect(toList());
+
+ assertThat(indexNames, contains(tableName + "_PK", "baz", "bar"));
+ }
+
+ @Test
+ void testReplaceIndexMissingIndex() {
+ CatalogSchemaDescriptor schema = catalogManager.activeSchema(DEFAULT_SCHEMA_NAME, clock.nowLong());
+
+ assertThat(schema, is(notNullValue()));
+
+ var index = mock(CatalogIndexDescriptor.class);
+
+ when(index.id()).thenReturn(Integer.MAX_VALUE);
+
+ Exception e = assertThrows(CatalogValidationException.class, () -> replaceIndex(schema, index));
+
+ assertThat(e.getMessage(), is(String.format("Index with ID %d has not been found in schema with ID %d", index.id(), 0)));
}
@Test
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/event/EventListener.java b/modules/core/src/main/java/org/apache/ignite/internal/event/EventListener.java
index 44491ea5dc..941b175636 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/event/EventListener.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/event/EventListener.java
@@ -17,7 +17,11 @@
package org.apache.ignite.internal.event;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture;
+
import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
import org.jetbrains.annotations.Nullable;
/** A listener that handles events from an event producer. */
@@ -33,4 +37,23 @@ public interface EventListener<P extends EventParameters> {
* future will never be completed with {@code null} value.
*/
CompletableFuture<Boolean> notify(P parameters, @Nullable Throwable exception);
+
+ /**
+ * Creates an adapter for a given callback.
+ *
+ * <p>Created listener will never return a future completed with {@code true}.
+ */
+ static <P extends EventParameters> EventListener<P> fromConsumer(Consumer<P> callback) {
+ return (parameters, exception) -> {
+ if (exception == null) {
+ try {
+ callback.accept(parameters);
+ } catch (Throwable e) {
+ return failedFuture(e);
+ }
+ }
+
+ return falseCompletedFuture();
+ };
+ }
}
diff --git a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
index 8ca12f33df..b3ff072ed1 100644
--- a/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
+++ b/modules/distribution-zones/src/main/java/org/apache/ignite/internal/distributionzones/DistributionZoneManager.java
@@ -61,7 +61,6 @@ import static org.apache.ignite.internal.util.ByteUtils.bytesToLong;
import static org.apache.ignite.internal.util.ByteUtils.fromBytes;
import static org.apache.ignite.internal.util.ByteUtils.longToBytes;
import static org.apache.ignite.internal.util.ByteUtils.toBytes;
-import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
@@ -1390,7 +1389,7 @@ public class DistributionZoneManager implements IgniteComponent {
catalogManager.listen(ZONE_DROP, (parameters, exception) -> inBusyLock(busyLock, () -> {
assert exception == null : parameters;
- return onDropZoneBusy((DropZoneEventParameters) parameters).thenCompose((ignored) -> falseCompletedFuture());
+ return onDropZoneBusy((DropZoneEventParameters) parameters).thenApply((ignored) -> false);
}));
catalogManager.listen(ZONE_ALTER, new ManagerCatalogAlterZoneEventListener());
diff --git a/modules/index/build.gradle b/modules/index/build.gradle
index 5df6d62100..381a2f8bf4 100644
--- a/modules/index/build.gradle
+++ b/modules/index/build.gradle
@@ -62,6 +62,7 @@ dependencies {
integrationTestImplementation project(':ignite-network-api')
integrationTestImplementation project(':ignite-replicator')
integrationTestImplementation project(':ignite-storage-api')
+ integrationTestImplementation project(':ignite-transactions')
integrationTestImplementation testFixtures(project(':ignite-core'))
integrationTestImplementation testFixtures(project(':ignite-runner'))
integrationTestImplementation testFixtures(project(':ignite-sql-engine'))
diff --git a/modules/index/src/integrationTest/java/org/apache/ignite/internal/index/ItBuildIndexTest.java b/modules/index/src/integrationTest/java/org/apache/ignite/internal/index/ItBuildIndexTest.java
index 8faadf6bb0..e4a054e262 100644
--- a/modules/index/src/integrationTest/java/org/apache/ignite/internal/index/ItBuildIndexTest.java
+++ b/modules/index/src/integrationTest/java/org/apache/ignite/internal/index/ItBuildIndexTest.java
@@ -17,16 +17,23 @@
package org.apache.ignite.internal.index;
+import static java.util.concurrent.CompletableFuture.runAsync;
import static java.util.stream.Collectors.joining;
import static org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus.AVAILABLE;
+import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.raft.util.OptimizedMarshaller.NO_POOL;
import static org.apache.ignite.internal.sql.engine.util.QueryChecker.containsIndexScan;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willTimeoutFast;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedFast;
+import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -38,6 +45,8 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.BiPredicate;
import java.util.stream.Stream;
@@ -45,9 +54,12 @@ import org.apache.ignite.Ignite;
import org.apache.ignite.internal.app.IgniteImpl;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
+import org.apache.ignite.internal.catalog.events.CatalogEvent;
+import org.apache.ignite.internal.catalog.events.RemoveIndexEventParameters;
+import org.apache.ignite.internal.catalog.events.StartBuildingIndexEventParameters;
import org.apache.ignite.internal.hlc.HybridClock;
-import org.apache.ignite.internal.lang.IgniteStringFormatter;
import org.apache.ignite.internal.network.NetworkMessage;
import org.apache.ignite.internal.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.internal.raft.Command;
@@ -60,9 +72,12 @@ import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.table.TableTestUtils;
import org.apache.ignite.internal.table.TableViewInternal;
import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand;
+import org.apache.ignite.internal.table.distributed.replication.request.BuildIndexReplicaRequest;
import org.apache.ignite.internal.table.distributed.schema.PartitionCommandsMarshallerImpl;
+import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.raft.jraft.rpc.WriteActionRequest;
import org.apache.ignite.table.Table;
+import org.apache.ignite.tx.TransactionOptions;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
@@ -105,7 +120,7 @@ public class ItBuildIndexTest extends BaseSqlIntegrationTest {
checkIndexBuild(partitions, replicas, INDEX_NAME);
- assertQuery(IgniteStringFormatter.format("SELECT * FROM {} WHERE i1 > 0", TABLE_NAME))
+ assertQuery(format("SELECT * FROM {} WHERE i1 > 0", TABLE_NAME))
.matches(containsIndexScan("PUBLIC", TABLE_NAME, INDEX_NAME))
.returns(1, 1)
.returns(2, 2)
@@ -115,6 +130,190 @@ public class ItBuildIndexTest extends BaseSqlIntegrationTest {
.check();
}
+ @Test
+ void testDropIndexDuringTransaction() throws Exception {
+ int partitions = initialNodes();
+
+ int replicas = initialNodes();
+
+ createAndPopulateTable(replicas, partitions);
+
+ createIndex(INDEX_NAME);
+
+ checkIndexBuild(partitions, replicas, INDEX_NAME);
+
+ CompletableFuture<Void> indexRemovedFuture = indexRemovedFuture();
+
+ IgniteImpl node = CLUSTER.aliveNode();
+
+ // Start a transaction. We expect that the index will not be removed until this transaction completes.
+ node.transactions().runInTransaction(tx -> {
+ dropIndex(INDEX_NAME);
+
+ CatalogIndexDescriptor indexDescriptor = getIndexDescriptor(node, INDEX_NAME);
+
+ assertThat(indexDescriptor, is(notNullValue()));
+ assertThat(indexDescriptor.status(), is(CatalogIndexStatus.STOPPING));
+ assertThat(indexRemovedFuture, willTimeoutFast());
+ }, new TransactionOptions().readOnly(false));
+
+ assertThat(indexRemovedFuture, willCompleteSuccessfully());
+ }
+
+ @Test
+ void testWritingIntoStoppingIndex() throws Exception {
+ int partitions = initialNodes();
+
+ int replicas = initialNodes();
+
+ createAndPopulateTable(replicas, partitions);
+
+ createIndex(INDEX_NAME);
+
+ checkIndexBuild(partitions, replicas, INDEX_NAME);
+
+ IgniteImpl node = CLUSTER.aliveNode();
+
+ // Latch for waiting for the RW transaction to start before dropping the index.
+ var startTransactionLatch = new CountDownLatch(1);
+ // Latch for waiting for the index to be dropped, before inserting data in the transaction.
+ var dropIndexLatch = new CountDownLatch(1);
+
+ CompletableFuture<Void> dropIndexFuture = runAsync(() -> {
+ try {
+ startTransactionLatch.await(1, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ throw new CompletionException(e);
+ }
+
+ dropIndex(INDEX_NAME);
+
+ CatalogIndexDescriptor indexDescriptor = getIndexDescriptor(node, INDEX_NAME);
+
+ assertThat(indexDescriptor, is(notNullValue()));
+ assertThat(indexDescriptor.status(), is(CatalogIndexStatus.STOPPING));
+
+ dropIndexLatch.countDown();
+ });
+
+ CompletableFuture<Void> insertDataIntoIndexTransaction = runAsync(() -> {
+ node.transactions().runInTransaction(tx -> {
+ startTransactionLatch.countDown();
+
+ try {
+ dropIndexLatch.await(1, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ throw new CompletionException(e);
+ }
+
+ // Insert data into a STOPPING index. We expect it to be inserted.
+ sql(tx, format("INSERT INTO {} VALUES {}", TABLE_NAME, toValuesString(List.of(239, 239))));
+
+ assertQuery((InternalTransaction) tx, format("SELECT * FROM {} WHERE i1 > 10", TABLE_NAME))
+ .matches(containsIndexScan("PUBLIC", TABLE_NAME, INDEX_NAME))
+ .returns(239, 239)
+ .check();
+ }, new TransactionOptions().readOnly(false));
+ });
+
+ assertThat(dropIndexFuture, willCompleteSuccessfully());
+ assertThat(insertDataIntoIndexTransaction, willCompleteSuccessfully());
+ }
+
+ @Test
+ void testDropIndexAfterRegistering() {
+ int partitions = initialNodes();
+
+ int replicas = initialNodes();
+
+ createAndPopulateTable(replicas, partitions);
+
+ CompletableFuture<Void> indexRemovedFuture = indexRemovedFuture();
+
+ CLUSTER.aliveNode().transactions().runInTransaction(tx -> {
+ // Create an index inside a transaction, this will prevent the index from building.
+ try {
+ createIndex(INDEX_NAME);
+
+ dropIndex(INDEX_NAME);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }, new TransactionOptions().readOnly(false));
+
+ assertThat(indexRemovedFuture, willCompleteSuccessfully());
+ }
+
+ @Test
+ void testDropIndexDuringBuilding() throws Exception {
+ int partitions = initialNodes();
+
+ int replicas = initialNodes();
+
+ createAndPopulateTable(replicas, partitions);
+
+ // Block index building messages, this way index will never become AVAILABLE.
+ CLUSTER.runningNodes().forEach(ignite -> ignite.dropMessages((id, message) -> message instanceof BuildIndexReplicaRequest));
+
+ CompletableFuture<Void> indexBuildingFuture = indexBuildingFuture();
+
+ CompletableFuture<Void> indexRemovedFuture = indexRemovedFuture();
+
+ createIndex(INDEX_NAME);
+
+ assertThat(indexBuildingFuture, willCompleteSuccessfully());
+
+ dropIndex(INDEX_NAME);
+
+ assertThat(indexRemovedFuture, willCompleteSuccessfully());
+ }
+
+ private static CompletableFuture<Void> indexBuildingFuture() {
+ IgniteImpl node = CLUSTER.aliveNode();
+
+ var indexBuildingFuture = new CompletableFuture<Void>();
+
+ node.catalogManager().listen(CatalogEvent.INDEX_BUILDING, (StartBuildingIndexEventParameters parameters, Throwable e) -> {
+ if (e == null) {
+ CatalogIndexDescriptor indexDescriptor = node.catalogManager().index(parameters.indexId(), parameters.catalogVersion());
+
+ if (indexDescriptor != null && indexDescriptor.name().equals(INDEX_NAME)) {
+ indexBuildingFuture.complete(null);
+ }
+ } else {
+ indexBuildingFuture.completeExceptionally(e);
+ }
+
+ return falseCompletedFuture();
+ });
+
+ return indexBuildingFuture;
+ }
+
+ private static CompletableFuture<Void> indexRemovedFuture() {
+ IgniteImpl node = CLUSTER.aliveNode();
+
+ var indexRemovedFuture = new CompletableFuture<Void>();
+
+ node.catalogManager().listen(CatalogEvent.INDEX_REMOVED, (RemoveIndexEventParameters parameters, Throwable e) -> {
+ if (e == null) {
+ node.catalogManager()
+ .catalog(parameters.catalogVersion() - 1)
+ .indexes()
+ .stream()
+ .filter(index -> index.name().equals(INDEX_NAME))
+ .findAny()
+ .ifPresent(index -> indexRemovedFuture.complete(null));
+ } else {
+ indexRemovedFuture.completeExceptionally(e);
+ }
+
+ return falseCompletedFuture();
+ });
+
+ return indexRemovedFuture;
+ }
+
@Test
@Disabled("https://issues.apache.org/jira/browse/IGNITE-20525")
void testChangePrimaryReplicaOnMiddleBuildIndex() throws Exception {
@@ -186,23 +385,23 @@ public class ItBuildIndexTest extends BaseSqlIntegrationTest {
}
private static void createAndPopulateTable(int replicas, int partitions) {
- sql(IgniteStringFormatter.format("CREATE ZONE IF NOT EXISTS {} WITH REPLICAS={}, PARTITIONS={}",
+ sql(format("CREATE ZONE IF NOT EXISTS {} WITH REPLICAS={}, PARTITIONS={}",
ZONE_NAME, replicas, partitions
));
- sql(IgniteStringFormatter.format(
+ sql(format(
"CREATE TABLE {} (i0 INTEGER PRIMARY KEY, i1 INTEGER) WITH PRIMARY_ZONE='{}'",
TABLE_NAME, ZONE_NAME
));
- sql(IgniteStringFormatter.format(
+ sql(format(
"INSERT INTO {} VALUES {}",
TABLE_NAME, toValuesString(List.of(1, 1), List.of(2, 2), List.of(3, 3), List.of(4, 4), List.of(5, 5))
));
}
private static void createIndex(String indexName) throws Exception {
- sql(IgniteStringFormatter.format("CREATE INDEX {} ON {} (i1)", indexName, TABLE_NAME));
+ sql(format("CREATE INDEX {} ON {} (i1)", indexName, TABLE_NAME));
waitForIndex(indexName);
}
@@ -291,7 +490,7 @@ public class ItBuildIndexTest extends BaseSqlIntegrationTest {
assertEquals(
replicas,
entry.getValue().size(),
- IgniteStringFormatter.format("p={}, nodes={}", entry.getKey(), entry.getValue())
+ format("p={}, nodes={}", entry.getKey(), entry.getValue())
);
}
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterTask.java b/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTask.java
similarity index 68%
rename from modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterTask.java
rename to modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTask.java
index 61313a4700..74f729be9e 100644
--- a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterTask.java
+++ b/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTask.java
@@ -33,6 +33,7 @@ import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeoutException;
@@ -40,15 +41,16 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import java.util.function.Supplier;
import org.apache.ignite.internal.catalog.Catalog;
+import org.apache.ignite.internal.catalog.CatalogCommand;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.ClockWaiter;
-import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologySnapshot;
import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.index.message.IndexMessagesFactory;
import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeRequest;
import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeResponse;
@@ -57,7 +59,6 @@ import org.apache.ignite.internal.lang.NodeStoppingException;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.network.ClusterService;
-import org.apache.ignite.internal.network.NetworkMessage;
import org.apache.ignite.internal.network.RecipientLeftException;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.PrimaryReplicaAwaitTimeoutException;
@@ -66,27 +67,36 @@ import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
/**
- * Task of switching the index from {@link CatalogIndexStatus#REGISTERED} to state {@link CatalogIndexStatus#BUILDING}, it is expected that
- * the task is executed on the node that is the primary replica for partition {@code 0} of the table for which the index is being created.
+ * Class representing an abstract task to move an index between different states.
*
- * <br><p>Approximate algorithm for the task:</p>
- * <ul>
- * <li>Waiting for the activation (on every cluster node) of the catalog version in which the index was created.</li>
- * <li>Make sure that the local node is still the primary replica for partition {@code 0} of the table.</li>
- * <li>Get the logical topology from the CMG leader.</li>
+ * <p>Such tasks are intended to be used in cases, when an index cannot be transferred to a new state by simply issuing a corresponding
+ * command. These cases include:
+ *
+ * <ol>
+ * <li>Changing the state of an index from {@link CatalogIndexStatus#REGISTERED} to {@link CatalogIndexStatus#BUILDING};</li>
+ * <li>Changing the state of an index from {@link CatalogIndexStatus#STOPPING} to {@code READ_ONLY} ({@code READ_ONLY} state is
+ * equivalent to the index being removed to the catalog).</li>
+ * </ol>
+ *
+ * <p>Approximate algorithm for a task:
+ * <ol>
+ * <li>Wait for the activation (on every cluster node) of the catalog version specified by
+ * {@link CatalogIndexDescriptor#txWaitCatalogVersion()};</li>
+ * <li>Make sure that the local node is still the primary replica for partition {@code 0} of the table that the index belongs to;</li>
+ * <li>Retrieve the logical topology from the CMG leader;</li>
* <li>For each node in the logical topology, send {@link IsNodeFinishedRwTransactionsStartedBeforeRequest} and process the following
* results: <ul>
- * <li>{@link IsNodeFinishedRwTransactionsStartedBeforeResponse#finished()} was {@code true}, finishing sending request.</li>
- * <li>{@link IsNodeFinishedRwTransactionsStartedBeforeResponse#finished()} was {@code false}, after a short interval send the
- * request again.</li>
- * <li>Node has left the physical topology, after a short interval send the request again.</li>
- * <li>Node has left the logical topology, finishing sending request.</li>
+ * <li>If {@link IsNodeFinishedRwTransactionsStartedBeforeResponse#finished()} is {@code true} - finish request processing;</li>
+ * <li>If {@link IsNodeFinishedRwTransactionsStartedBeforeResponse#finished()} is {@code false} - send the request again after a
+ * short period of time;</li>
+ * <li>If the target node has left the physical topology - send the request again after a short period of time;</li>
+ * <li>If the target node has left the logical topology - finish request processing.</li>
* </ul></li>
- * <li>Change the index status to {@link CatalogIndexStatus#BUILDING}.</li>
- * </ul>
+ * <li>Execute the Catalog command to change the state of the index.</li>
+ * </ol>
*/
-class IndexBuildingStarterTask {
- private static final IgniteLogger LOG = Loggers.forClass(IndexBuildingStarterTask.class);
+abstract class ChangeIndexStatusTask {
+ private static final IgniteLogger LOG = Loggers.forClass(ChangeIndexStatusTask.class);
private static final IndexMessagesFactory FACTORY = new IndexMessagesFactory();
@@ -116,7 +126,7 @@ class IndexBuildingStarterTask {
private final AtomicBoolean taskStopGuard = new AtomicBoolean();
- IndexBuildingStarterTask(
+ ChangeIndexStatusTask(
CatalogIndexDescriptor indexDescriptor,
CatalogManager catalogManager,
PlacementDriver placementDriver,
@@ -149,28 +159,33 @@ class IndexBuildingStarterTask {
}
try {
- return supplyAsync(() -> awaitActivateForCatalogVersionOfIndexCreation()
- .thenCompose(unused -> ensureThatLocalNodeStillPrimaryReplica())
- .thenCompose(unused -> inBusyLocks(logicalTopologyService::logicalTopologyOnLeader))
- .thenComposeAsync(this::awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation, executor)
- .thenComposeAsync(unused -> switchIndexToBuildingStatus(), executor), executor)
- .thenCompose(Function.identity())
- .handle((unused, throwable) -> {
+ LOG.debug("Starting task to change index status. Index: {}", indexDescriptor);
+
+ return awaitCatalogVersionActivation()
+ .thenComposeAsync(unused -> ensureThatLocalNodeStillPrimaryReplica(), executor)
+ .thenComposeAsync(unused -> inBusyLocks(logicalTopologyService::logicalTopologyOnLeader), executor)
+ .thenComposeAsync(this::awaitFinishRwTxsBeforeCatalogVersion, executor)
+ .thenComposeAsync(unused -> inBusyLocks(() -> catalogManager.execute(switchIndexStatusCommand())), executor)
+ .whenComplete((unused, throwable) -> {
if (throwable != null) {
Throwable cause = unwrapCause(throwable);
if (!(cause instanceof IndexTaskStoppingException) && !(cause instanceof NodeStoppingException)) {
- LOG.error("Error starting index building: {}", cause, indexDescriptor.id());
+ LOG.error("Error starting index task: {}", cause, indexDescriptor.id());
}
}
-
- return null;
});
} finally {
leaveBusy();
}
}
+ /**
+ * Returns a CatalogCommand that will be submitted to the Catalog after all RW transactions that had observed the Catalog at
+ * {@link CatalogIndexDescriptor#txWaitCatalogVersion()} have finished their execution.
+ */
+ abstract CatalogCommand switchIndexStatusCommand();
+
/** Stops the execution of a task. */
void stop() {
if (!taskStopGuard.compareAndSet(false, true)) {
@@ -180,15 +195,29 @@ class IndexBuildingStarterTask {
taskBusyLock.block();
}
- private CompletableFuture<Void> awaitActivateForCatalogVersionOfIndexCreation() {
- return inBusyLocks(() -> {
- Catalog catalog = catalogManager.catalog(indexDescriptor.creationCatalogVersion());
+ CatalogIndexDescriptor targetIndex() {
+ return indexDescriptor;
+ }
- assert catalog != null : IgniteStringFormatter.format("Missing catalog version: [indexId={}, catalogVersion={}]",
- indexDescriptor.id(), indexDescriptor.creationCatalogVersion());
+ private CompletableFuture<Void> awaitCatalogVersionActivation() {
+ CompletableFuture<HybridTimestamp> activationTimestampFuture = supplyAsync(() -> {
+ if (!enterBusy()) {
+ throw new IndexTaskStoppingException();
+ }
- return clockWaiter.waitFor(clusterWideEnsuredActivationTimestamp(catalog));
- });
+ try {
+ Catalog catalog = catalogManager.catalog(indexDescriptor.txWaitCatalogVersion());
+
+ assert catalog != null : IgniteStringFormatter.format("Missing catalog version: [indexId={}, catalogVersion={}]",
+ indexDescriptor.id(), indexDescriptor.txWaitCatalogVersion());
+
+ return clusterWideEnsuredActivationTimestamp(catalog);
+ } finally {
+ leaveBusy();
+ }
+ }, executor);
+
+ return activationTimestampFuture.thenCompose(clockWaiter::waitFor);
}
private CompletableFuture<Void> ensureThatLocalNodeStillPrimaryReplica() {
@@ -229,7 +258,7 @@ class IndexBuildingStarterTask {
});
}
- private CompletableFuture<Void> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(LogicalTopologySnapshot logicalTopologySnapshot) {
+ private CompletableFuture<Void> awaitFinishRwTxsBeforeCatalogVersion(LogicalTopologySnapshot logicalTopologySnapshot) {
return inBusyLocks(() -> {
Set<LogicalNode> remainingNodes = ConcurrentHashMap.newKeySet();
remainingNodes.addAll(logicalTopologySnapshot.nodes());
@@ -237,8 +266,8 @@ class IndexBuildingStarterTask {
var nodeLeftLogicalTopologyListener = new NodeLeftLogicalTopologyListener(remainingNodes);
logicalTopologyService.addEventListener(nodeLeftLogicalTopologyListener);
- CompletableFuture[] futures = remainingNodes.stream()
- .map(node -> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(node, remainingNodes))
+ CompletableFuture<?>[] futures = remainingNodes.stream()
+ .map(node -> awaitFinishRwTxsBeforeCatalogVersion(node, remainingNodes))
.toArray(CompletableFuture[]::new);
return allOf(futures)
@@ -246,10 +275,7 @@ class IndexBuildingStarterTask {
});
}
- private CompletableFuture<NetworkMessage> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(
- LogicalNode targetNode,
- Set<LogicalNode> remainingNodes
- ) {
+ private CompletableFuture<Void> awaitFinishRwTxsBeforeCatalogVersion(LogicalNode targetNode, Set<LogicalNode> remainingNodes) {
return inBusyLocks(() -> {
if (!remainingNodes.contains(targetNode)) {
// Target node left logical topology, there is no need to do anything.
@@ -259,17 +285,22 @@ class IndexBuildingStarterTask {
IsNodeFinishedRwTransactionsStartedBeforeRequest request = isNodeFinishedRwTransactionsStartedBeforeRequest();
return clusterService.messagingService().invoke(targetNode, request, INVOKE_MESSAGE_TIMEOUT_MILLS)
- .handle((message, throwable) -> {
- if (throwable != null) {
- Throwable cause = unwrapCause(throwable);
+ .exceptionally(throwable -> {
+ Throwable cause = unwrapCause(throwable);
+
+ if (!(cause instanceof TimeoutException) && !(cause instanceof RecipientLeftException)) {
+ throw new CompletionException(cause);
+ }
+
+ return null;
+ })
+ .thenCompose(message -> {
+ if (message != null && ((IsNodeFinishedRwTransactionsStartedBeforeResponse) message).finished()) {
+ LOG.debug("All transactions have finished on node {}. Remaining nodes: {}", targetNode, remainingNodes);
- if (!(cause instanceof TimeoutException) && !(cause instanceof RecipientLeftException)) {
- return CompletableFuture.<NetworkMessage>failedFuture(cause);
- }
- } else if (((IsNodeFinishedRwTransactionsStartedBeforeResponse) message).finished()) {
remainingNodes.remove(targetNode);
- return completedFuture(message);
+ return nullCompletedFuture();
}
// Let's try to send the message again, but after a short time.
@@ -277,21 +308,17 @@ class IndexBuildingStarterTask {
Executor delayedExecutor = delayedExecutor(RETRY_SEND_MESSAGE_TIMEOUT_MILLS, MILLISECONDS, executor);
return supplyAsync(
- () -> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(targetNode, remainingNodes),
+ () -> awaitFinishRwTxsBeforeCatalogVersion(targetNode, remainingNodes),
delayedExecutor
).thenCompose(Function.identity());
});
- }).thenCompose(Function.identity());
+ });
});
}
- private CompletableFuture<Void> switchIndexToBuildingStatus() {
- return inBusyLocks(() -> catalogManager.execute(StartBuildingIndexCommand.builder().indexId(indexDescriptor.id()).build()));
- }
-
private IsNodeFinishedRwTransactionsStartedBeforeRequest isNodeFinishedRwTransactionsStartedBeforeRequest() {
return FACTORY.isNodeFinishedRwTransactionsStartedBeforeRequest()
- .targetCatalogVersion(indexDescriptor.creationCatalogVersion())
+ .targetCatalogVersion(indexDescriptor.txWaitCatalogVersion())
.build();
}
@@ -304,14 +331,14 @@ class IndexBuildingStarterTask {
}
private <T> CompletableFuture<T> inBusyLocks(Supplier<CompletableFuture<T>> supplier) {
- if (!IndexManagementUtils.enterBusy(busyLock, taskBusyLock)) {
+ if (!enterBusy()) {
return failedFuture(new IndexTaskStoppingException());
}
try {
return supplier.get();
} finally {
- IndexManagementUtils.leaveBusy(busyLock, taskBusyLock);
+ leaveBusy();
}
}
}
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterController.java b/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskController.java
similarity index 55%
rename from modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterController.java
rename to modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskController.java
index 16dc4ebd16..4787cb31d6 100644
--- a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterController.java
+++ b/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskController.java
@@ -17,22 +17,22 @@
package org.apache.ignite.internal.index;
-import static java.util.concurrent.CompletableFuture.failedFuture;
-import static org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus.REGISTERED;
import static org.apache.ignite.internal.index.IndexManagementUtils.isLocalNode;
-import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
-import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import java.util.Set;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.CatalogService;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
import org.apache.ignite.internal.catalog.events.CatalogEvent;
import org.apache.ignite.internal.catalog.events.CreateIndexEventParameters;
import org.apache.ignite.internal.catalog.events.RemoveIndexEventParameters;
+import org.apache.ignite.internal.catalog.events.StoppingIndexEventParameters;
import org.apache.ignite.internal.close.ManuallyCloseable;
+import org.apache.ignite.internal.event.EventListener;
import org.apache.ignite.internal.network.ClusterService;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
@@ -42,30 +42,36 @@ import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
/**
- * Component is responsible for starting and stopping {@link IndexBuildingStarterTask} via {@link IndexBuildingStarter}.
+ * Component that reacts to certain Catalog changes and starts or stops corresponding {@link ChangeIndexStatusTask}s via the
+ * {@link ChangeIndexStatusTaskScheduler}.
*
- * <p>Tasks will be started only if the local node is the primary replica for the {@code 0} partition of the table for which indexes are
- * created.</p>
+ * <p>Tasks will be started only if the local node is the primary replica for the {@code 0} partition of the table which indexes have been
+ * modified in the Catalog.
*
- * <br><p>Tasks are started and stopped based on events:</p>
+ * <p>The following events are being monitored:
* <ul>
- * <li>{@link CatalogEvent#INDEX_CREATE} - task starts for the new index.</li>
- * <li>{@link CatalogEvent#INDEX_REMOVED} - stops task for the new index if it has been added.</li>
- * <li>{@link PrimaryReplicaEvent#PRIMARY_REPLICA_ELECTED} - when the local node becomes the primary replica for partition {@code 0} of
- * the table, it starts tasks for its all new indexes, and when the primary replica expires, it stops all tasks for the table.</li>
+ * <li>{@link CatalogEvent#INDEX_CREATE} - when an index is created, a task will be started to move the index to the
+ * {@link CatalogIndexStatus#BUILDING} state; </li>
+ * <li>{@link CatalogEvent#INDEX_STOPPING} - when an index is dropped, a task will be started to remove the index from the Catalog;</li>
+ * <li>{@link CatalogEvent#INDEX_REMOVED} - when an index is removed from the Catalog, all ongoing tasks for the index will be stopped;
+ * </li>
+ * <li>{@link PrimaryReplicaEvent#PRIMARY_REPLICA_ELECTED} - if the local node has become the primary replica for partition {@code 0} of
+ * the table, it starts corresponding tasks depending on the current status of indices in the Catalog. If the local node stops being the
+ * primary replica for the partition {@code 0}, it stops all tasks that belong to indices of the table that the partition belongs to.
+ * </li>
* </ul>
*
- * <br><p>On node recovery, tasks will be started on {@link PrimaryReplicaEvent#PRIMARY_REPLICA_ELECTED}, which will fire due to a change in
- * the {@link ReplicaMeta#getLeaseholderId()} on node restart but after {@link ReplicaMeta#getExpirationTime()}.</p>
+ * <p>On node recovery, tasks will be started on {@link PrimaryReplicaEvent#PRIMARY_REPLICA_ELECTED}, which will fire due to a change in
+ * the {@link ReplicaMeta#getLeaseholderId()} on node restart but after {@link ReplicaMeta#getExpirationTime()}.
*/
-class IndexBuildingStarterController implements ManuallyCloseable {
+class ChangeIndexStatusTaskController implements ManuallyCloseable {
private final CatalogService catalogService;
private final PlacementDriver placementDriver;
private final ClusterService clusterService;
- private final IndexBuildingStarter indexBuildingStarter;
+ private final ChangeIndexStatusTaskScheduler changeIndexStatusTaskScheduler;
/** Tables IDs for which the local node is the primary replica for the partition with ID {@code 0}. */
private final Set<Integer> localNodeIsPrimaryReplicaForTableIds = ConcurrentHashMap.newKeySet();
@@ -74,16 +80,16 @@ class IndexBuildingStarterController implements ManuallyCloseable {
private final AtomicBoolean closeGuard = new AtomicBoolean();
- IndexBuildingStarterController(
- CatalogService catalogService,
+ ChangeIndexStatusTaskController(
+ CatalogManager catalogManager,
PlacementDriver placementDriver,
ClusterService clusterService,
- IndexBuildingStarter indexBuildingStarter
+ ChangeIndexStatusTaskScheduler changeIndexStatusTaskScheduler
) {
- this.catalogService = catalogService;
+ this.catalogService = catalogManager;
this.placementDriver = placementDriver;
this.clusterService = clusterService;
- this.indexBuildingStarter = indexBuildingStarter;
+ this.changeIndexStatusTaskScheduler = changeIndexStatusTaskScheduler;
addListeners();
}
@@ -96,68 +102,56 @@ class IndexBuildingStarterController implements ManuallyCloseable {
busyLock.block();
- indexBuildingStarter.close();
+ changeIndexStatusTaskScheduler.close();
}
private void addListeners() {
- catalogService.listen(CatalogEvent.INDEX_CREATE, (parameters, exception) -> {
- if (exception != null) {
- return failedFuture(exception);
- }
-
- return onIndexCreate((CreateIndexEventParameters) parameters).thenApply(unused -> false);
- });
-
- catalogService.listen(CatalogEvent.INDEX_REMOVED, (parameters, exception) -> {
- if (exception != null) {
- return failedFuture(exception);
- }
+ catalogService.listen(CatalogEvent.INDEX_CREATE, EventListener.fromConsumer(this::onIndexCreated));
- return onIndexRemoved((RemoveIndexEventParameters) parameters).thenApply(unused -> false);
- });
+ catalogService.listen(CatalogEvent.INDEX_STOPPING, EventListener.fromConsumer(this::onIndexDropped));
- placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, (parameters, exception) -> {
- if (exception != null) {
- return failedFuture(exception);
- }
+ catalogService.listen(CatalogEvent.INDEX_REMOVED, EventListener.fromConsumer(this::onIndexRemoved));
- return onPrimaryReplicaElected(parameters).thenApply(unused -> false);
- });
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, EventListener.fromConsumer(this::onPrimaryReplicaElected));
}
- private CompletableFuture<?> onIndexCreate(CreateIndexEventParameters parameters) {
- return inBusyLockAsync(busyLock, () -> {
+ private void onIndexCreated(CreateIndexEventParameters parameters) {
+ inBusyLock(busyLock, () -> {
CatalogIndexDescriptor indexDescriptor = parameters.indexDescriptor();
if (localNodeIsPrimaryReplicaForTableIds.contains(indexDescriptor.tableId())) {
// Schedule building the index only if the local node is the primary replica for the 0 partition of the table for which the
// index was created.
- indexBuildingStarter.scheduleTask(parameters.indexDescriptor());
+ changeIndexStatusTaskScheduler.scheduleStartBuildingTask(parameters.indexDescriptor());
}
-
- return nullCompletedFuture();
});
}
- private CompletableFuture<?> onIndexRemoved(RemoveIndexEventParameters parameters) {
- return inBusyLockAsync(busyLock, () -> {
- CatalogIndexDescriptor indexDescriptor = catalogService.index(parameters.indexId(), parameters.catalogVersion() - 1);
+ private void onIndexDropped(StoppingIndexEventParameters parameters) {
+ inBusyLock(busyLock, () -> {
+ CatalogIndexDescriptor indexDescriptor = catalogService.index(parameters.indexId(), parameters.catalogVersion());
assert indexDescriptor != null : parameters.indexId();
- indexBuildingStarter.stopTask(indexDescriptor);
-
- return nullCompletedFuture();
+ if (localNodeIsPrimaryReplicaForTableIds.contains(indexDescriptor.tableId())) {
+ // Schedule index removal only if the local node is the primary replica for the 0 partition of the table for which the
+ // index was dropped.
+ changeIndexStatusTaskScheduler.scheduleRemoveIndexTask(indexDescriptor);
+ }
});
}
- private CompletableFuture<?> onPrimaryReplicaElected(PrimaryReplicaEventParameters parameters) {
- return inBusyLockAsync(busyLock, () -> {
+ private void onIndexRemoved(RemoveIndexEventParameters parameters) {
+ inBusyLock(busyLock, () -> changeIndexStatusTaskScheduler.stopStartBuildingTask(parameters.indexId()));
+ }
+
+ private void onPrimaryReplicaElected(PrimaryReplicaEventParameters parameters) {
+ inBusyLock(busyLock, () -> {
TablePartitionId primaryReplicaId = (TablePartitionId) parameters.groupId();
if (primaryReplicaId.partitionId() != 0) {
// We are only interested in the 0 partition.
- return nullCompletedFuture();
+ return;
}
int tableId = primaryReplicaId.tableId();
@@ -168,11 +162,9 @@ class IndexBuildingStarterController implements ManuallyCloseable {
}
} else {
if (localNodeIsPrimaryReplicaForTableIds.remove(tableId)) {
- indexBuildingStarter.stopTasks(tableId);
+ changeIndexStatusTaskScheduler.stopTasksForTable(tableId);
}
}
-
- return nullCompletedFuture();
});
}
@@ -181,8 +173,19 @@ class IndexBuildingStarterController implements ManuallyCloseable {
int catalogVersion = catalogService.latestCatalogVersion();
for (CatalogIndexDescriptor indexDescriptor : catalogService.indexes(catalogVersion, tableId)) {
- if (indexDescriptor.status() == REGISTERED) {
- indexBuildingStarter.scheduleTask(indexDescriptor);
+ switch (indexDescriptor.status()) {
+ case REGISTERED:
+ changeIndexStatusTaskScheduler.scheduleStartBuildingTask(indexDescriptor);
+
+ break;
+
+ case STOPPING:
+ changeIndexStatusTaskScheduler.scheduleRemoveIndexTask(indexDescriptor);
+
+ break;
+
+ default:
+ break;
}
}
}
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterTaskId.java b/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskId.java
similarity index 61%
rename from modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterTaskId.java
rename to modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskId.java
index 917ce69110..929a787df1 100644
--- a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterTaskId.java
+++ b/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskId.java
@@ -17,25 +17,24 @@
package org.apache.ignite.internal.index;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
import org.apache.ignite.internal.tostring.S;
-/** {@link IndexBuildingStarterTask} ID. */
-class IndexBuildingStarterTaskId {
- private final int tableId;
-
+/** {@link ChangeIndexStatusTask} ID. */
+class ChangeIndexStatusTaskId {
private final int indexId;
- IndexBuildingStarterTaskId(int tableId, int indexId) {
- this.tableId = tableId;
- this.indexId = indexId;
- }
+ private final CatalogIndexStatus status;
- public int tableId() {
- return tableId;
+ ChangeIndexStatusTaskId(int indexId, CatalogIndexStatus status) {
+ this.indexId = indexId;
+ this.status = status;
}
- public int indexId() {
- return indexId;
+ ChangeIndexStatusTaskId(CatalogIndexDescriptor indexDescriptor) {
+ this.indexId = indexDescriptor.id();
+ this.status = indexDescriptor.status();
}
@Override
@@ -47,20 +46,20 @@ class IndexBuildingStarterTaskId {
return false;
}
- IndexBuildingStarterTaskId that = (IndexBuildingStarterTaskId) o;
+ ChangeIndexStatusTaskId that = (ChangeIndexStatusTaskId) o;
- return tableId == that.tableId && indexId == that.indexId;
+ return indexId == that.indexId && status == that.status;
}
@Override
public int hashCode() {
- int result = tableId;
- result = 31 * result + indexId;
+ int result = indexId;
+ result = 31 * result + status.hashCode();
return result;
}
@Override
public String toString() {
- return S.toString(IndexBuildingStarterTaskId.class, this);
+ return S.toString(ChangeIndexStatusTaskId.class, this);
}
}
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskScheduler.java b/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskScheduler.java
new file mode 100644
index 0000000000..c0342e0313
--- /dev/null
+++ b/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskScheduler.java
@@ -0,0 +1,213 @@
+/*
+ * 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.ignite.internal.index;
+
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockSafe;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.ignite.internal.catalog.CatalogCommand;
+import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.ClockWaiter;
+import org.apache.ignite.internal.catalog.commands.RemoveIndexCommand;
+import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
+import org.apache.ignite.internal.close.ManuallyCloseable;
+import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.network.ClusterService;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+
+/** Component is responsible for starting and stopping {@link ChangeIndexStatusTask}. */
+class ChangeIndexStatusTaskScheduler implements ManuallyCloseable {
+ private static final IgniteLogger LOG = Loggers.forClass(ChangeIndexStatusTaskScheduler.class);
+
+ private final CatalogManager catalogManager;
+
+ private final ClusterService clusterService;
+
+ private final LogicalTopologyService logicalTopologyService;
+
+ private final HybridClock clock;
+
+ private final ClockWaiter clockWaiter;
+
+ private final PlacementDriver placementDriver;
+
+ private final Executor executor;
+
+ private final Map<ChangeIndexStatusTaskId, ChangeIndexStatusTask> taskById = new ConcurrentHashMap<>();
+
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ private final AtomicBoolean closeGuard = new AtomicBoolean();
+
+ ChangeIndexStatusTaskScheduler(
+ CatalogManager catalogManager,
+ ClusterService clusterService,
+ LogicalTopologyService logicalTopologyService,
+ HybridClock clock,
+ ClockWaiter clockWaiter,
+ PlacementDriver placementDriver,
+ Executor executor
+ ) {
+ this.catalogManager = catalogManager;
+ this.clusterService = clusterService;
+ this.logicalTopologyService = logicalTopologyService;
+ this.clock = clock;
+ this.clockWaiter = clockWaiter;
+ this.placementDriver = placementDriver;
+ this.executor = executor;
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (!closeGuard.compareAndSet(false, true)) {
+ return;
+ }
+
+ busyLock.block();
+
+ stopAllTasks();
+ }
+
+ /**
+ * Schedules a task for that will transfer the given index to the {@link CatalogIndexStatus#BUILDING} state.
+ */
+ void scheduleStartBuildingTask(CatalogIndexDescriptor indexDescriptor) {
+ assert indexDescriptor.status() == CatalogIndexStatus.REGISTERED;
+
+ LOG.info("Scheduling starting of index building. Index: {}", indexDescriptor);
+
+ inBusyLockSafe(busyLock, () -> {
+ var taskId = new ChangeIndexStatusTaskId(indexDescriptor);
+
+ scheduleTaskBusy(taskId, startBuildingIndexTask(indexDescriptor));
+ });
+ }
+
+ private ChangeIndexStatusTask startBuildingIndexTask(CatalogIndexDescriptor indexDescriptor) {
+ return new ChangeIndexStatusTask(
+ indexDescriptor,
+ catalogManager,
+ placementDriver,
+ clusterService,
+ logicalTopologyService,
+ clock,
+ clockWaiter,
+ executor,
+ busyLock
+ ) {
+ @Override
+ CatalogCommand switchIndexStatusCommand() {
+ return StartBuildingIndexCommand.builder().indexId(indexDescriptor.id()).build();
+ }
+ };
+ }
+
+ /**
+ * Schedules a task for that will remove a given index from the Catalog.
+ */
+ void scheduleRemoveIndexTask(CatalogIndexDescriptor indexDescriptor) {
+ assert indexDescriptor.status() == CatalogIndexStatus.STOPPING;
+
+ LOG.info("Scheduling index removal. Index: {}", indexDescriptor);
+
+ inBusyLockSafe(busyLock, () -> {
+ var taskId = new ChangeIndexStatusTaskId(indexDescriptor);
+
+ scheduleTaskBusy(taskId, removeIndexTask(indexDescriptor));
+ });
+ }
+
+ private ChangeIndexStatusTask removeIndexTask(CatalogIndexDescriptor indexDescriptor) {
+ return new ChangeIndexStatusTask(
+ indexDescriptor,
+ catalogManager,
+ placementDriver,
+ clusterService,
+ logicalTopologyService,
+ clock,
+ clockWaiter,
+ executor,
+ busyLock
+ ) {
+ @Override
+ CatalogCommand switchIndexStatusCommand() {
+ return RemoveIndexCommand.builder().indexId(indexDescriptor.id()).build();
+ }
+ };
+ }
+
+ private void scheduleTaskBusy(ChangeIndexStatusTaskId taskId, ChangeIndexStatusTask task) {
+ // Check if the task has already been added before.
+ if (taskById.putIfAbsent(taskId, task) == null) {
+ task.start().whenComplete((unused, throwable) -> taskById.remove(taskId));
+ } else {
+ LOG.info("Skipping task scheduling, because a task with the same ID is already running.");
+ }
+ }
+
+ /**
+ * Stops an ongoing {@link ChangeIndexStatusTask} for a given index if it is present.
+ *
+ * @param indexId Index ID.
+ */
+ void stopStartBuildingTask(int indexId) {
+ inBusyLockSafe(busyLock, () -> {
+ ChangeIndexStatusTask removed = taskById.remove(new ChangeIndexStatusTaskId(indexId, CatalogIndexStatus.REGISTERED));
+
+ if (removed != null) {
+ removed.stop();
+ }
+ });
+ }
+
+ /**
+ * Stops {@link ChangeIndexStatusTask}s that correspond to all indices of a given table.
+ *
+ * @param tableId Table ID.
+ */
+ void stopTasksForTable(int tableId) {
+ inBusyLockSafe(busyLock, () -> {
+ Iterator<ChangeIndexStatusTask> it = taskById.values().iterator();
+
+ while (it.hasNext()) {
+ ChangeIndexStatusTask task = it.next();
+
+ if (task.targetIndex().tableId() == tableId) {
+ it.remove();
+
+ task.stop();
+ }
+ }
+ });
+ }
+
+ private void stopAllTasks() {
+ taskById.values().forEach(ChangeIndexStatusTask::stop);
+ taskById.clear();
+ }
+}
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildController.java b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildController.java
index 436bfa67ea..c45459dc66 100644
--- a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildController.java
+++ b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildController.java
@@ -312,7 +312,7 @@ class IndexBuildController implements ManuallyCloseable {
mvPartition,
localNode(),
enlistmentConsistencyToken,
- indexDescriptor.creationCatalogVersion()
+ indexDescriptor.txWaitCatalogVersion()
);
}
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingManager.java b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingManager.java
index f57f4949ee..fe04152b14 100644
--- a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingManager.java
+++ b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingManager.java
@@ -60,13 +60,11 @@ public class IndexBuildingManager implements IgniteComponent {
private final IndexBuilder indexBuilder;
- private final IndexBuildingStarter indexBuildingStarter;
-
private final IndexAvailabilityController indexAvailabilityController;
private final IndexBuildController indexBuildController;
- private final IndexBuildingStarterController indexBuildingStarterController;
+ private final ChangeIndexStatusTaskController changeIndexStatusTaskController;
private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
@@ -102,7 +100,11 @@ public class IndexBuildingManager implements IgniteComponent {
indexBuilder = new IndexBuilder(executor, replicaService);
- indexBuildingStarter = new IndexBuildingStarter(
+ indexAvailabilityController = new IndexAvailabilityController(catalogManager, metaStorageManager, indexBuilder);
+
+ indexBuildController = new IndexBuildController(indexBuilder, indexManager, catalogManager, clusterService, placementDriver, clock);
+
+ var indexTaskScheduler = new ChangeIndexStatusTaskScheduler(
catalogManager,
clusterService,
logicalTopologyService,
@@ -112,15 +114,11 @@ public class IndexBuildingManager implements IgniteComponent {
executor
);
- indexAvailabilityController = new IndexAvailabilityController(catalogManager, metaStorageManager, indexBuilder);
-
- indexBuildController = new IndexBuildController(indexBuilder, indexManager, catalogManager, clusterService, placementDriver, clock);
-
- indexBuildingStarterController = new IndexBuildingStarterController(
+ changeIndexStatusTaskController = new ChangeIndexStatusTaskController(
catalogManager,
placementDriver,
clusterService,
- indexBuildingStarter
+ indexTaskScheduler
);
}
@@ -149,10 +147,9 @@ public class IndexBuildingManager implements IgniteComponent {
closeAllManually(
indexBuilder,
- indexBuildingStarter,
indexAvailabilityController,
indexBuildController,
- indexBuildingStarterController
+ changeIndexStatusTaskController
);
shutdownAndAwaitTermination(executor, 10, TimeUnit.SECONDS);
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarter.java b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarter.java
deleted file mode 100644
index 301e9a02da..0000000000
--- a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarter.java
+++ /dev/null
@@ -1,163 +0,0 @@
-/*
- * 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.ignite.internal.index;
-
-import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockSafe;
-
-import java.util.Iterator;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executor;
-import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.ignite.internal.catalog.CatalogManager;
-import org.apache.ignite.internal.catalog.ClockWaiter;
-import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
-import org.apache.ignite.internal.close.ManuallyCloseable;
-import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
-import org.apache.ignite.internal.hlc.HybridClock;
-import org.apache.ignite.internal.network.ClusterService;
-import org.apache.ignite.internal.placementdriver.PlacementDriver;
-import org.apache.ignite.internal.util.IgniteSpinBusyLock;
-
-/** Component is responsible for starting and stopping {@link IndexBuildingStarterTask}. */
-class IndexBuildingStarter implements ManuallyCloseable {
- private final CatalogManager catalogManager;
-
- private final ClusterService clusterService;
-
- private final LogicalTopologyService logicalTopologyService;
-
- private final HybridClock clock;
-
- private final ClockWaiter clockWaiter;
-
- private final PlacementDriver placementDriver;
-
- private final Executor executor;
-
- private final Map<IndexBuildingStarterTaskId, IndexBuildingStarterTask> taskById = new ConcurrentHashMap<>();
-
- private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
-
- private final AtomicBoolean closeGuard = new AtomicBoolean();
-
- IndexBuildingStarter(
- CatalogManager catalogManager,
- ClusterService clusterService,
- LogicalTopologyService logicalTopologyService,
- HybridClock clock,
- ClockWaiter clockWaiter,
- PlacementDriver placementDriver,
- Executor executor
- ) {
- this.catalogManager = catalogManager;
- this.clusterService = clusterService;
- this.logicalTopologyService = logicalTopologyService;
- this.clock = clock;
- this.clockWaiter = clockWaiter;
- this.placementDriver = placementDriver;
- this.executor = executor;
- }
-
- @Override
- public void close() throws Exception {
- if (!closeGuard.compareAndSet(false, true)) {
- return;
- }
-
- busyLock.block();
-
- stopAllTasks();
- }
-
- /**
- * Schedules {@link IndexBuildingStarterTask} for the table index if it is not already in progress.
- *
- * @param indexDescriptor Index descriptor.
- */
- void scheduleTask(CatalogIndexDescriptor indexDescriptor) {
- inBusyLockSafe(busyLock, () -> {
- IndexBuildingStarterTaskId taskId = taskId(indexDescriptor);
-
- IndexBuildingStarterTask task = new IndexBuildingStarterTask(
- indexDescriptor,
- catalogManager,
- placementDriver,
- clusterService,
- logicalTopologyService,
- clock,
- clockWaiter,
- executor,
- busyLock
- );
-
- if (taskById.putIfAbsent(taskId, task) != null) {
- // Task has already been added and is running.
- return;
- }
-
- task.start().whenComplete((unused, throwable) -> taskById.remove(taskId));
- });
- }
-
- /**
- * Stops {@link IndexBuildingStarterTask} for the table index if it is present.
- *
- * @param indexDescriptor Index descriptor.
- */
- void stopTask(CatalogIndexDescriptor indexDescriptor) {
- inBusyLockSafe(busyLock, () -> {
- IndexBuildingStarterTask removed = taskById.remove(taskId(indexDescriptor));
-
- if (removed != null) {
- removed.stop();
- }
- });
- }
-
- /**
- * Stops all {@link IndexBuildingStarterTask} that are present for the table.
- *
- * @param tableId Table ID.
- */
- void stopTasks(int tableId) {
- inBusyLockSafe(busyLock, () -> {
- Iterator<Entry<IndexBuildingStarterTaskId, IndexBuildingStarterTask>> it = taskById.entrySet().iterator();
-
- while (it.hasNext()) {
- Entry<IndexBuildingStarterTaskId, IndexBuildingStarterTask> e = it.next();
-
- if (e.getKey().tableId() == tableId) {
- it.remove();
-
- e.getValue().stop();
- }
- }
- });
- }
-
- private void stopAllTasks() {
- taskById.values().forEach(IndexBuildingStarterTask::stop);
- taskById.clear();
- }
-
- private static IndexBuildingStarterTaskId taskId(CatalogIndexDescriptor indexDescriptor) {
- return new IndexBuildingStarterTaskId(indexDescriptor.tableId(), indexDescriptor.id());
- }
-}
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexChooser.java b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexChooser.java
index 3247cf3dc6..35224e5bfd 100644
--- a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexChooser.java
+++ b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexChooser.java
@@ -20,19 +20,15 @@ package org.apache.ignite.internal.index;
import static java.util.Collections.binarySearch;
import static java.util.Collections.unmodifiableList;
import static java.util.Comparator.comparingInt;
-import static java.util.concurrent.CompletableFuture.failedFuture;
import static org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus.AVAILABLE;
import static org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus.STOPPING;
-import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
-import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map.Entry;
import java.util.NavigableMap;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.ignite.internal.catalog.CatalogService;
@@ -43,6 +39,7 @@ import org.apache.ignite.internal.catalog.events.CatalogEvent;
import org.apache.ignite.internal.catalog.events.DropTableEventParameters;
import org.apache.ignite.internal.catalog.events.RemoveIndexEventParameters;
import org.apache.ignite.internal.close.ManuallyCloseable;
+import org.apache.ignite.internal.event.EventListener;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
/** Index chooser for various operations, for example for RW transactions. */
@@ -175,25 +172,13 @@ class IndexChooser implements ManuallyCloseable {
}
private void addListeners() {
- catalogService.listen(CatalogEvent.INDEX_REMOVED, (parameters, exception) -> {
- if (exception != null) {
- return failedFuture(exception);
- }
-
- return onIndexRemoved((RemoveIndexEventParameters) parameters).thenApply(unused -> false);
- });
-
- catalogService.listen(CatalogEvent.TABLE_DROP, (parameters, exception) -> {
- if (exception != null) {
- return failedFuture(exception);
- }
+ catalogService.listen(CatalogEvent.INDEX_REMOVED, EventListener.fromConsumer(this::onIndexRemoved));
- return onDropTable((DropTableEventParameters) parameters).thenApply(unused -> false);
- });
+ catalogService.listen(CatalogEvent.TABLE_DROP, EventListener.fromConsumer(this::onDropTable));
}
- private CompletableFuture<?> onIndexRemoved(RemoveIndexEventParameters parameters) {
- return inBusyLockAsync(busyLock, () -> {
+ private void onIndexRemoved(RemoveIndexEventParameters parameters) {
+ inBusyLock(busyLock, () -> {
int previousCatalogVersion = parameters.catalogVersion() - 1;
CatalogIndexDescriptor removedIndexDescriptor = catalogService.index(parameters.indexId(), previousCatalogVersion);
@@ -201,22 +186,18 @@ class IndexChooser implements ManuallyCloseable {
assert removedIndexDescriptor != null : "indexId=" + parameters.indexId() + ", catalogVersion=" + previousCatalogVersion;
if (removedIndexDescriptor.status() != AVAILABLE && removedIndexDescriptor.status() != STOPPING) {
- return nullCompletedFuture();
+ return;
}
addRemovedAvailableIndex(removedIndexDescriptor, parameters.catalogVersion());
-
- return nullCompletedFuture();
});
}
- private CompletableFuture<?> onDropTable(DropTableEventParameters parameters) {
- return inBusyLockAsync(busyLock, () -> {
+ private void onDropTable(DropTableEventParameters parameters) {
+ inBusyLock(busyLock, () -> {
// We can remove removed indexes on table drop as we need such indexes only for writing, and write operations will be denied
// right after a table drop has been activated.
removedAvailableTableIndexes.entrySet().removeIf(entry -> parameters.tableId() == entry.getKey().tableId);
-
- return nullCompletedFuture();
});
}
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java
index 0d234bdc37..6815e18ada 100644
--- a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java
+++ b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java
@@ -21,8 +21,6 @@ import static java.util.concurrent.CompletableFuture.allOf;
import static java.util.concurrent.CompletableFuture.failedFuture;
import static java.util.stream.Collectors.toMap;
import static org.apache.ignite.internal.catalog.events.CatalogEvent.INDEX_CREATE;
-import static org.apache.ignite.internal.catalog.events.CatalogEvent.INDEX_STOPPING;
-import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
@@ -42,11 +40,8 @@ import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.LongFunction;
-import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.CatalogService;
-import org.apache.ignite.internal.catalog.commands.RemoveIndexCommand;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
-import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
import org.apache.ignite.internal.catalog.events.CatalogEvent;
import org.apache.ignite.internal.catalog.events.CreateIndexEventParameters;
@@ -90,9 +85,8 @@ public class IndexManager implements IgniteComponent {
/** Table manager. */
private final TableManager tableManager;
- // TODO: IGNITE-21117 -change the field type to CatalogService again.
/** Catalog service. */
- private final CatalogManager catalogManager;
+ private final CatalogService catalogService;
/** Meta storage manager. */
private final MetaStorageManager metaStorageManager;
@@ -119,20 +113,20 @@ public class IndexManager implements IgniteComponent {
*
* @param schemaManager Schema manager.
* @param tableManager Table manager.
- * @param catalogManager Catalog manager.
+ * @param catalogService Catalog service.
* @param ioExecutor Separate executor for IO operations like storage initialization.
*/
public IndexManager(
SchemaManager schemaManager,
TableManager tableManager,
- CatalogManager catalogManager,
+ CatalogService catalogService,
MetaStorageManager metaStorageManager,
ExecutorService ioExecutor,
Consumer<LongFunction<CompletableFuture<?>>> registry
) {
this.schemaManager = schemaManager;
this.tableManager = tableManager;
- this.catalogManager = catalogManager;
+ this.catalogService = catalogService;
this.metaStorageManager = metaStorageManager;
this.ioExecutor = ioExecutor;
@@ -146,7 +140,7 @@ public class IndexManager implements IgniteComponent {
startIndexes();
- catalogManager.listen(INDEX_CREATE, (parameters, exception) -> {
+ catalogService.listen(INDEX_CREATE, (parameters, exception) -> {
if (exception != null) {
return failedFuture(exception);
}
@@ -154,31 +148,11 @@ public class IndexManager implements IgniteComponent {
return onIndexCreate((CreateIndexEventParameters) parameters);
});
- // TODO: IGNITE-21117 - remove this.
- catalogManager.listen(INDEX_STOPPING, (parameters, exception) -> {
- if (exception != null) {
- return failedFuture(exception);
- }
-
- removeIndex(((StoppingIndexEventParameters) parameters).indexId());
-
- return falseCompletedFuture();
- });
-
LOG.info("Index manager started");
return nullCompletedFuture();
}
- private CompletableFuture<Void> removeIndex(int indexId) {
- return catalogManager.execute(RemoveIndexCommand.builder().indexId(indexId).build())
- .whenComplete((res, ex) -> {
- if (ex != null) {
- LOG.error("Cannot remove a dropped index [indexId={}]", ex, indexId);
- }
- });
- }
-
@Override
public void stop() throws Exception {
LOG.debug("Index manager is about to stop");
@@ -246,7 +220,7 @@ public class IndexManager implements IgniteComponent {
long causalityToken = parameters.causalityToken();
int catalogVersion = parameters.catalogVersion();
- CatalogTableDescriptor table = catalogManager.table(tableId, catalogVersion);
+ CatalogTableDescriptor table = catalogService.table(tableId, catalogVersion);
assert table != null : "tableId=" + tableId + ", indexId=" + indexId;
@@ -355,17 +329,11 @@ public class IndexManager implements IgniteComponent {
List<CompletableFuture<?>> startIndexFutures = new ArrayList<>();
- Map<CatalogTableDescriptor, Collection<CatalogIndexDescriptor>> indexesForRecovery = collectIndexesForRecovery(catalogManager);
+ Map<CatalogTableDescriptor, Collection<CatalogIndexDescriptor>> indexesForRecovery = collectIndexesForRecovery(catalogService);
for (Entry<CatalogTableDescriptor, Collection<CatalogIndexDescriptor>> e : indexesForRecovery.entrySet()) {
CatalogTableDescriptor table = e.getKey();
for (CatalogIndexDescriptor index : e.getValue()) {
- // TODO: IGNITE-21117 - remove this.
- if (index.status() == CatalogIndexStatus.STOPPING
- && catalogManager.index(index.id(), catalogManager.latestCatalogVersion()) != null) {
- startIndexFutures.add(removeIndex(index.id()));
- }
-
startIndexFutures.add(startIndexAsync(table, index, causalityToken));
}
}
diff --git a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildingStarterControllerTest.java b/modules/index/src/test/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskControllerTest.java
similarity index 72%
rename from modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildingStarterControllerTest.java
rename to modules/index/src/test/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskControllerTest.java
index 31162fcba3..c8c06da286 100644
--- a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildingStarterControllerTest.java
+++ b/modules/index/src/test/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskControllerTest.java
@@ -25,19 +25,22 @@ import static org.apache.ignite.internal.index.TestIndexManagementUtils.NODE_ID;
import static org.apache.ignite.internal.index.TestIndexManagementUtils.NODE_NAME;
import static org.apache.ignite.internal.index.TestIndexManagementUtils.TABLE_NAME;
import static org.apache.ignite.internal.index.TestIndexManagementUtils.createTable;
-import static org.apache.ignite.internal.index.TestIndexManagementUtils.dropIndex;
import static org.apache.ignite.internal.index.TestIndexManagementUtils.newPrimaryReplicaMeta;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.util.List;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.CatalogTestUtils;
+import org.apache.ignite.internal.catalog.commands.MakeIndexAvailableCommand;
+import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridClockImpl;
@@ -46,7 +49,7 @@ import org.apache.ignite.internal.network.ClusterNodeImpl;
import org.apache.ignite.internal.network.ClusterService;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
import org.apache.ignite.internal.replicator.TablePartitionId;
-import org.apache.ignite.internal.testframework.IgniteAbstractTest;
+import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.NetworkAddress;
@@ -59,9 +62,9 @@ import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
-/** For {@link IndexBuildingStarterController} testing. */
+/** For {@link ChangeIndexStatusTaskController} testing. */
@ExtendWith(MockitoExtension.class)
-public class IndexBuildingStarterControllerTest extends IgniteAbstractTest {
+public class ChangeIndexStatusTaskControllerTest extends BaseIgniteAbstractTest {
private final HybridClock clock = new HybridClockImpl();
private final CatalogManager catalogManager = CatalogTestUtils.createTestCatalogManager(NODE_NAME, clock);
@@ -71,9 +74,9 @@ public class IndexBuildingStarterControllerTest extends IgniteAbstractTest {
private final ClusterService clusterService = createClusterService();
@Mock
- private IndexBuildingStarter indexBuildingStarter;
+ private ChangeIndexStatusTaskScheduler changeIndexStatusTaskScheduler;
- private @Nullable IndexBuildingStarterController controller;
+ private @Nullable ChangeIndexStatusTaskController taskController;
@BeforeEach
void setUp() {
@@ -81,15 +84,17 @@ public class IndexBuildingStarterControllerTest extends IgniteAbstractTest {
createTable(catalogManager, TABLE_NAME, COLUMN_NAME);
- controller = new IndexBuildingStarterController(catalogManager, placementDriver, clusterService, indexBuildingStarter);
+ taskController = new ChangeIndexStatusTaskController(
+ catalogManager, placementDriver, clusterService, changeIndexStatusTaskScheduler
+ );
}
@AfterEach
void tearDown() throws Exception {
- IgniteUtils.closeAll(
+ IgniteUtils.closeAllManually(
catalogManager::beforeNodeStop,
catalogManager::stop,
- controller == null ? null : controller::close
+ taskController
);
}
@@ -97,11 +102,36 @@ public class IndexBuildingStarterControllerTest extends IgniteAbstractTest {
void testStartTaskOnIndexCreateEvent() {
setPrimaryReplicaLocalNode();
- verify(indexBuildingStarter, never()).scheduleTask(any());
+ verify(changeIndexStatusTaskScheduler, never()).scheduleStartBuildingTask(any());
createIndex();
- verify(indexBuildingStarter).scheduleTask(eq(indexDescriptor()));
+ verify(changeIndexStatusTaskScheduler).scheduleStartBuildingTask(eq(indexDescriptor()));
+ }
+
+ @Test
+ void testStartTaskOnIndexDropEvent() {
+ setPrimaryReplicaLocalNode();
+
+ createIndex();
+
+ CatalogIndexDescriptor indexDescriptor = indexDescriptor();
+
+ int indexId = indexDescriptor.id();
+
+ assertThat(
+ catalogManager.execute(List.of(
+ StartBuildingIndexCommand.builder().indexId(indexId).build(),
+ MakeIndexAvailableCommand.builder().indexId(indexId).build())
+ ),
+ willCompleteSuccessfully()
+ );
+
+ verify(changeIndexStatusTaskScheduler, never()).scheduleRemoveIndexTask(any());
+
+ dropIndex();
+
+ verify(changeIndexStatusTaskScheduler).scheduleRemoveIndexTask(argThat(descriptor -> descriptor.id() == indexId));
}
@Test
@@ -112,9 +142,9 @@ public class IndexBuildingStarterControllerTest extends IgniteAbstractTest {
CatalogIndexDescriptor indexDescriptor = indexDescriptor();
- dropIndex(catalogManager, INDEX_NAME);
+ dropIndex();
- verify(indexBuildingStarter).stopTask(eq(indexDescriptor));
+ verify(changeIndexStatusTaskScheduler).stopStartBuildingTask(eq(indexDescriptor.id()));
}
@Test
@@ -123,11 +153,11 @@ public class IndexBuildingStarterControllerTest extends IgniteAbstractTest {
CatalogIndexDescriptor indexDescriptor = indexDescriptor();
- verify(indexBuildingStarter, never()).scheduleTask(eq(indexDescriptor));
+ verify(changeIndexStatusTaskScheduler, never()).scheduleStartBuildingTask(eq(indexDescriptor));
setPrimaryReplicaLocalNode();
- verify(indexBuildingStarter).scheduleTask(eq(indexDescriptor));
+ verify(changeIndexStatusTaskScheduler).scheduleStartBuildingTask(eq(indexDescriptor));
}
@Test
@@ -138,13 +168,17 @@ public class IndexBuildingStarterControllerTest extends IgniteAbstractTest {
setPrimaryReplicaAnotherNode();
- verify(indexBuildingStarter).stopTasks(eq(tableId()));
+ verify(changeIndexStatusTaskScheduler).stopTasksForTable(eq(tableId()));
}
private void createIndex() {
TestIndexManagementUtils.createIndex(catalogManager, TABLE_NAME, INDEX_NAME, COLUMN_NAME);
}
+ private void dropIndex() {
+ TestIndexManagementUtils.dropIndex(catalogManager, INDEX_NAME);
+ }
+
private void setPrimaryReplicaLocalNode() {
setPrimaryReplica(LOCAL_NODE);
}
diff --git a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildingStarterTaskTest.java b/modules/index/src/test/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskTest.java
similarity index 93%
rename from modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildingStarterTaskTest.java
rename to modules/index/src/test/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskTest.java
index 0f473a3c0d..11b7c04217 100644
--- a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildingStarterTaskTest.java
+++ b/modules/index/src/test/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskTest.java
@@ -36,6 +36,7 @@ import static org.apache.ignite.internal.index.TestIndexManagementUtils.createTa
import static org.apache.ignite.internal.index.TestIndexManagementUtils.indexDescriptor;
import static org.apache.ignite.internal.index.TestIndexManagementUtils.newPrimaryReplicaMeta;
import static org.apache.ignite.internal.index.TestIndexManagementUtils.startBuildingIndex;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrow;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
@@ -57,8 +58,11 @@ import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeoutException;
+import org.apache.ignite.internal.catalog.CatalogCommand;
import org.apache.ignite.internal.catalog.CatalogManager;
+import org.apache.ignite.internal.catalog.ChangeIndexStatusValidationException;
import org.apache.ignite.internal.catalog.ClockWaiter;
+import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyEventListener;
@@ -89,9 +93,9 @@ import org.mockito.Mock;
import org.mockito.Spy;
import org.mockito.junit.jupiter.MockitoExtension;
-/** For {@link IndexBuildingStarterTask} testing. */
+/** For {@link ChangeIndexStatusTask} testing. */
@ExtendWith(MockitoExtension.class)
-public class IndexBuildingStarterTaskTest extends IgniteAbstractTest {
+public class ChangeIndexStatusTaskTest extends IgniteAbstractTest {
private static final IndexMessagesFactory FACTORY = new IndexMessagesFactory();
private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
@@ -113,7 +117,7 @@ public class IndexBuildingStarterTaskTest extends IgniteAbstractTest {
@Mock(strictness = LENIENT)
private LogicalTopologyService logicalTopologyService;
- private IndexBuildingStarterTask task;
+ private ChangeIndexStatusTask task;
private CatalogIndexDescriptor indexDescriptor;
@@ -140,7 +144,7 @@ public class IndexBuildingStarterTaskTest extends IgniteAbstractTest {
when(logicalTopologyService.logicalTopologyOnLeader()).thenReturn(logicalTopologySnapshotFuture);
- task = new IndexBuildingStarterTask(
+ task = new ChangeIndexStatusTask(
indexDescriptor,
catalogManager,
placementDriver,
@@ -150,7 +154,12 @@ public class IndexBuildingStarterTaskTest extends IgniteAbstractTest {
clockWaiter,
executor,
busyLock
- );
+ ) {
+ @Override
+ CatalogCommand switchIndexStatusCommand() {
+ return StartBuildingIndexCommand.builder().indexId(indexDescriptor.id()).build();
+ }
+ };
}
@AfterEach
@@ -212,7 +221,7 @@ public class IndexBuildingStarterTaskTest extends IgniteAbstractTest {
awaitPrimaryReplicaFuture1
);
- assertThat(task.start(), willCompleteSuccessfully());
+ assertThat(task.start(), willThrow(IndexTaskStoppingException.class));
assertEquals(REGISTERED, actualIndexStatus());
verify(placementDriver, times(2)).awaitPrimaryReplica(any(), any(), anyLong(), any());
@@ -229,7 +238,7 @@ public class IndexBuildingStarterTaskTest extends IgniteAbstractTest {
awaitPrimaryReplicaFuture1
);
- assertThat(task.start(), willCompleteSuccessfully());
+ assertThat(task.start(), willThrow(PrimaryReplicaAwaitException.class));
assertEquals(REGISTERED, actualIndexStatus());
verify(placementDriver, times(2)).awaitPrimaryReplica(any(), any(), anyLong(), any());
@@ -285,13 +294,12 @@ public class IndexBuildingStarterTaskTest extends IgniteAbstractTest {
@Test
void testFailedSendIsNodeFinishedRwTransactionsStartedBeforeRequest() {
- CompletableFuture<NetworkMessage> invokeFuture = failedFuture(new Exception());
-
- when(clusterService.messagingService().invoke(any(ClusterNode.class), any(), anyLong())).thenReturn(invokeFuture);
+ when(clusterService.messagingService().invoke(any(ClusterNode.class), any(), anyLong()))
+ .thenReturn(failedFuture(new Exception("test")));
clearInvocations(clockWaiter);
- assertThat(task.start(), willCompleteSuccessfully());
+ assertThat(task.start(), willThrow(Exception.class, "test"));
assertEquals(REGISTERED, actualIndexStatus());
verify(clusterService.messagingService(), times(1)).invoke(any(ClusterNode.class), any(), anyLong());
@@ -302,7 +310,7 @@ public class IndexBuildingStarterTaskTest extends IgniteAbstractTest {
void testIndexAlreadyInBuildingStatus() {
startBuildingIndex(catalogManager, indexDescriptor.id());
- assertThat(task.start(), willCompleteSuccessfully());
+ assertThat(task.start(), willThrow(ChangeIndexStatusValidationException.class));
assertEquals(BUILDING, actualIndexStatus());
}
diff --git a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexAvailabilityControllerTest.java b/modules/index/src/test/java/org/apache/ignite/internal/index/IndexAvailabilityControllerTest.java
index 2e36085313..f9922a67d3 100644
--- a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexAvailabilityControllerTest.java
+++ b/modules/index/src/test/java/org/apache/ignite/internal/index/IndexAvailabilityControllerTest.java
@@ -434,6 +434,6 @@ public class IndexAvailabilityControllerTest extends BaseIgniteAbstractTest {
}
private int indexCreationCatalogVersion(String indexName) {
- return getIndexStrict(catalogManager, indexName, clock.nowLong()).creationCatalogVersion();
+ return getIndexStrict(catalogManager, indexName, clock.nowLong()).txWaitCatalogVersion();
}
}
diff --git a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildControllerTest.java b/modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildControllerTest.java
index 4d7e197b47..50e24669e1 100644
--- a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildControllerTest.java
+++ b/modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildControllerTest.java
@@ -305,6 +305,6 @@ public class IndexBuildControllerTest extends BaseIgniteAbstractTest {
}
private int indexCreationCatalogVersion(String indexName) {
- return getIndexStrict(catalogManager, indexName, clock.nowLong()).creationCatalogVersion();
+ return getIndexStrict(catalogManager, indexName, clock.nowLong()).txWaitCatalogVersion();
}
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
index 8c77e17dbb..9d6f92f105 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
@@ -3772,7 +3772,7 @@ public class PartitionReplicaListener implements ReplicaListener {
CatalogIndexDescriptor indexDescriptor = latestIndexDescriptorInBuildingStatus(catalogService, tableId());
if (indexDescriptor != null) {
- txRwOperationTracker.updateMinAllowedCatalogVersionForStartOperation(indexDescriptor.creationCatalogVersion());
+ txRwOperationTracker.updateMinAllowedCatalogVersionForStartOperation(indexDescriptor.txWaitCatalogVersion());
}
catalogService.listen(CatalogEvent.INDEX_BUILDING, indexBuildingCatalogEventListener);
@@ -3795,7 +3795,7 @@ public class PartitionReplicaListener implements ReplicaListener {
assert indexDescriptor != null : "indexId=" + indexId + ", catalogVersion=" + parameters.catalogVersion();
if (indexDescriptor.tableId() == tableId()) {
- txRwOperationTracker.updateMinAllowedCatalogVersionForStartOperation(indexDescriptor.creationCatalogVersion());
+ txRwOperationTracker.updateMinAllowedCatalogVersionForStartOperation(indexDescriptor.txWaitCatalogVersion());
}
return falseCompletedFuture();
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
index 0b0c67e94c..67e40a345a 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
@@ -2881,7 +2881,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
when(hashIndexDescriptor.id()).thenReturn(indexId);
when(hashIndexDescriptor.tableId()).thenReturn(TABLE_ID);
when(hashIndexDescriptor.status()).thenReturn(BUILDING);
- when(hashIndexDescriptor.creationCatalogVersion()).thenReturn(1);
+ when(hashIndexDescriptor.txWaitCatalogVersion()).thenReturn(1);
when(catalogService.index(eq(indexId), anyInt())).thenReturn(hashIndexDescriptor);