You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "tkalkirill (via GitHub)" <gi...@apache.org> on 2024/01/26 05:00:53 UTC
[PR] IGNITE-21115 Wait for completion of transactions started before index appearance [ignite-3]
tkalkirill opened a new pull request, #3095:
URL: https://github.com/apache/ignite-3/pull/3095
https://issues.apache.org/jira/browse/IGNITE-21115
Thank you for submitting the pull request.
To streamline the review process of the patch and ensure better code quality
we ask both an author and a reviewer to verify the following:
### The Review Checklist
- [ ] **Formal criteria:** TC status, codestyle, mandatory documentation. Also make sure to complete the following:
\- There is a single JIRA ticket related to the pull request.
\- The web-link to the pull request is attached to the JIRA ticket.
\- The JIRA ticket has the Patch Available state.
\- The description of the JIRA ticket explains WHAT was made, WHY and HOW.
\- The pull request title is treated as the final commit message. The following pattern must be used: IGNITE-XXXX Change summary where XXXX - number of JIRA issue.
- [ ] **Design:** new code conforms with the design principles of the components it is added to.
- [ ] **Patch quality:** patch cannot be split into smaller pieces, its size must be reasonable.
- [ ] **Code quality:** code is clean and readable, necessary developer documentation is added if needed.
- [ ] **Tests code quality:** test set covers positive/negative scenarios, happy/edge cases. Tests are effective in terms of execution time and resources.
### Notes
- [Apache Ignite Coding Guidelines](https://cwiki.apache.org/confluence/display/IGNITE/Java+Code+Style+Guide)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-21115 Wait for completion of transactions started before index appearance [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #3095:
URL: https://github.com/apache/ignite-3/pull/3095#discussion_r1470630864
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterTask.java:
##########
@@ -0,0 +1,321 @@
+/*
+ * 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 java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.CompletableFuture.supplyAsync;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.catalog.commands.CatalogUtils.clusterWideEnsuredActivationTimestamp;
+import static org.apache.ignite.internal.index.IndexManagementUtils.AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC;
+import static org.apache.ignite.internal.index.IndexManagementUtils.earliestCatalogVersionOfIndexInRegisteredStatus;
+import static org.apache.ignite.internal.index.IndexManagementUtils.isPrimaryReplica;
+import static org.apache.ignite.internal.index.IndexManagementUtils.localNode;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeoutException;
+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.CatalogManager;
+import org.apache.ignite.internal.catalog.ClockWaiter;
+import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
+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.index.message.IndexMessagesFactory;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeRequest;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeResponse;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+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.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.PrimaryReplicaAwaitTimeoutException;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.RecipientLeftException;
+
+/**
+ * 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.
+ *
+ * <br><p>Approximate algorithm for the task:</p>
+ * <ul>
+ * <li>Waiting for the activation of the catalog version in which the index was created.</li>
Review Comment:
apply it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-21115 Wait for completion of transactions started before index appearance [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #3095:
URL: https://github.com/apache/ignite-3/pull/3095#discussion_r1470629299
##########
modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java:
##########
@@ -495,4 +496,18 @@ public static Collection<CatalogIndexDescriptor> collectIndexes(
return indexByIdMap.values();
}
+
+ /**
+ * Returns the activation timestamp of the catalog version for the whole cluster.
Review Comment:
apply it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-21115 Wait for completion of transactions started before index appearance [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #3095:
URL: https://github.com/apache/ignite-3/pull/3095#discussion_r1469146301
##########
modules/catalog/src/main/java/org/apache/ignite/internal/catalog/commands/CatalogUtils.java:
##########
@@ -495,4 +496,18 @@ public static Collection<CatalogIndexDescriptor> collectIndexes(
return indexByIdMap.values();
}
+
+ /**
+ * Returns the activation timestamp of the catalog version for the whole cluster.
Review Comment:
```suggestion
* Returns the timestamp at which the catalog version is guaranteed to be activated on every node of the cluster. This takes into account possible clock skew between nodes.
```
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterController.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * 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 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 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.CatalogService;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
+import org.apache.ignite.internal.catalog.events.CatalogEvent;
+import org.apache.ignite.internal.catalog.events.CreateIndexEventParameters;
+import org.apache.ignite.internal.catalog.events.DropIndexEventParameters;
+import org.apache.ignite.internal.close.ManuallyCloseable;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * Component is responsible for starting and stopping {@link IndexBuildingStarterTask} via {@link IndexBuildingStarter}.
+ *
+ * <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>
+ *
+ * <br><p>Tasks are started and stopped based on events:</p>
+ * <ul>
+ * <li>{@link CatalogEvent#INDEX_CREATE} - task starts for the new index.</li>
+ * <li>{@link CatalogEvent#INDEX_DROP} - 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 it all new indexes, and when the primary replica expires, it stops all tasks for the table.</li>
Review Comment:
```suggestion
* the table, it starts tasks for its all new indexes, and when the primary replica expires, it stops all tasks for the table.</li>
```
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarter.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.close.ManuallyCloseable;
+import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+
+/** 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 tableId Table ID.
+ * @param indexId Index ID.
+ */
+ void scheduleTask(int tableId, int indexId) {
+ inBusyLockSafe(busyLock, () -> {
+ IndexBuildingStarterTaskId taskId = new IndexBuildingStarterTaskId(tableId, indexId);
+
+ IndexBuildingStarterTask task = new IndexBuildingStarterTask(
+ indexId,
+ tableId,
+ 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));
Review Comment:
How about also logging an exception if it happens during start in another `whenComplete()`?
##########
modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildingStarterTaskTest.java:
##########
@@ -0,0 +1,349 @@
+/*
+ * 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 java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.Executors.newSingleThreadExecutor;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.catalog.CatalogTestUtils.createTestCatalogManager;
+import static org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus.BUILDING;
+import static org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus.REGISTERED;
+import static org.apache.ignite.internal.index.TestIndexManagementUtils.COLUMN_NAME;
+import static org.apache.ignite.internal.index.TestIndexManagementUtils.INDEX_NAME;
+import static org.apache.ignite.internal.index.TestIndexManagementUtils.LOCAL_NODE;
+import static org.apache.ignite.internal.index.TestIndexManagementUtils.LOGICAL_LOCAL_NODE;
+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.createIndex;
+import static org.apache.ignite.internal.index.TestIndexManagementUtils.createTable;
+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.CompletableFutureMatcher.willCompleteSuccessfully;
+import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
+import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mock.Strictness.LENIENT;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+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.CatalogManager;
+import org.apache.ignite.internal.catalog.ClockWaiter;
+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;
+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.HybridClockImpl;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.index.message.IndexMessagesFactory;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.PrimaryReplicaAwaitException;
+import org.apache.ignite.internal.placementdriver.PrimaryReplicaAwaitTimeoutException;
+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.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.MessagingService;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.RecipientLeftException;
+import org.apache.ignite.network.TopologyService;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+/** For {@link IndexBuildingStarterTask} testing. */
+@ExtendWith(MockitoExtension.class)
+public class IndexBuildingStarterTaskTest extends IgniteAbstractTest {
+ private static final IndexMessagesFactory FACTORY = new IndexMessagesFactory();
+
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ private final HybridClock clock = new HybridClockImpl();
+
+ private final ClockWaiter clockWaiter = spy(new ClockWaiter(NODE_NAME, clock));
+
+ private final CatalogManager catalogManager = createTestCatalogManager(NODE_NAME, clockWaiter);
+
+ private final ExecutorService executor = spy(newSingleThreadExecutor());
+
+ private final ClusterService clusterService = createClusterService();
+
+ @Mock
+ private PlacementDriver placementDriver;
+
+ @Mock(strictness = LENIENT)
+ private LogicalTopologyService logicalTopologyService;
+
+ private IndexBuildingStarterTask task;
+
+ private CatalogIndexDescriptor indexDescriptor;
+
+ @BeforeEach
+ void setUp() {
+ assertThat(allOf(clockWaiter.start(), catalogManager.start()), willCompleteSuccessfully());
+
+ createTable(catalogManager, TABLE_NAME, COLUMN_NAME);
+ createIndex(catalogManager, TABLE_NAME, INDEX_NAME, COLUMN_NAME);
+
+ indexDescriptor = indexDescriptor(catalogManager, INDEX_NAME, clock);
+
+ CompletableFuture<ReplicaMeta> localNodeReplicaMetaFuture = completedFuture(
+ createLocalNodeReplicaMeta(HybridTimestamp.MIN_VALUE, HybridTimestamp.MAX_VALUE)
+ );
+
+ when(placementDriver.awaitPrimaryReplica(any(), any(), anyLong(), any())).thenReturn(localNodeReplicaMetaFuture);
+
+ CompletableFuture<LogicalTopologySnapshot> logicalTopologySnapshotFuture = completedFuture(
+ new LogicalTopologySnapshot(1, List.of(LOGICAL_LOCAL_NODE))
+ );
+
+ when(logicalTopologyService.logicalTopologyOnLeader()).thenReturn(logicalTopologySnapshotFuture);
+
+ task = new IndexBuildingStarterTask(
+ indexDescriptor.id(),
+ indexDescriptor.tableId(),
+ catalogManager,
+ placementDriver,
+ clusterService,
+ logicalTopologyService,
+ clock,
+ clockWaiter,
+ executor,
+ busyLock
+ );
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ closeAll(
+ catalogManager::beforeNodeStop,
+ clockWaiter::beforeNodeStop,
+ catalogManager::stop,
+ clockWaiter::stop,
+ task == null ? null : task::stop
+ );
+
+ shutdownAndAwaitTermination(executor, 1, SECONDS);
+ }
+
+ @Test
+ void testSimpleTaskExecution() {
+ clearInvocations(clockWaiter);
Review Comment:
This should be done automatically by `MockitoExtension`, so this seems redundant. Or maybe you need to annotate the field with `@Spy` (and remove explicit `spy(`)
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterController.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * 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 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 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.CatalogService;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
+import org.apache.ignite.internal.catalog.events.CatalogEvent;
+import org.apache.ignite.internal.catalog.events.CreateIndexEventParameters;
+import org.apache.ignite.internal.catalog.events.DropIndexEventParameters;
+import org.apache.ignite.internal.close.ManuallyCloseable;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * Component is responsible for starting and stopping {@link IndexBuildingStarterTask} via {@link IndexBuildingStarter}.
+ *
+ * <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>
+ *
+ * <br><p>Tasks are started and stopped based on events:</p>
+ * <ul>
+ * <li>{@link CatalogEvent#INDEX_CREATE} - task starts for the new index.</li>
+ * <li>{@link CatalogEvent#INDEX_DROP} - 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 it all new indexes, and when the primary replica expires, it stops all tasks for the table.</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>
+ */
+class IndexBuildingStarterController implements ManuallyCloseable {
+ private final CatalogService catalogService;
+
+ private final PlacementDriver placementDriver;
+
+ private final ClusterService clusterService;
+
+ private final IndexBuildingStarter indexBuildingStarter;
+
+ /** Tables IDs for which the local node is the primary replica for the partition with ID {@code 0}. */
+ private final Set<Integer> tableIds = ConcurrentHashMap.newKeySet();
+
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ private final AtomicBoolean closeGuard = new AtomicBoolean();
+
+ IndexBuildingStarterController(
+ CatalogService catalogService,
+ PlacementDriver placementDriver,
+ ClusterService clusterService,
+ IndexBuildingStarter indexBuildingStarter
+ ) {
+ this.catalogService = catalogService;
+ this.placementDriver = placementDriver;
+ this.clusterService = clusterService;
+ this.indexBuildingStarter = indexBuildingStarter;
+
+ addListeners();
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (!closeGuard.compareAndSet(false, true)) {
+ return;
+ }
+
+ busyLock.block();
+
+ indexBuildingStarter.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_DROP, (parameters, exception) -> {
+ if (exception != null) {
+ return failedFuture(exception);
+ }
+
+ return onIndexDrop((DropIndexEventParameters) parameters).thenApply(unused -> false);
+ });
+
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, (parameters, exception) -> {
+ if (exception != null) {
+ return failedFuture(exception);
+ }
+
+ return onPrimaryReplicaElected(parameters).thenApply(unused -> false);
+ });
+ }
+
+ private CompletableFuture<?> onIndexCreate(CreateIndexEventParameters parameters) {
+ return inBusyLockAsync(busyLock, () -> {
+ CatalogIndexDescriptor indexDescriptor = parameters.indexDescriptor();
+
+ if (tableIds.contains(indexDescriptor.tableId())) {
+ // Start building the index only if the local node is the primary replica for the 0 partition of the table for which the
Review Comment:
```suggestion
// Start watiing before building the index only if the local node is the primary replica for the 0 partition of the table for which the
```
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingManager.java:
##########
@@ -79,16 +80,45 @@ public IndexBuildingManager(
IndexManager indexManager,
PlacementDriver placementDriver,
ClusterService clusterService,
- HybridClock clock
+ LogicalTopologyService logicalTopologyService,
+ HybridClock clock,
+ ClockWaiter clockWaiter
) {
this.metaStorageManager = metaStorageManager;
- this.catalogManager = catalogManager;
- indexBuilder = new IndexBuilder(nodeName, Runtime.getRuntime().availableProcessors(), replicaService);
+ int threadCount = Runtime.getRuntime().availableProcessors();
+
+ executor = new ThreadPoolExecutor(
+ threadCount,
Review Comment:
Do we really need to keep all these threads alive all the time? It seems that a node needs to build indexes only in some periods of time, but most of time it doesn't need these threads. Should `coreThreads` be set to 0?
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterTask.java:
##########
@@ -0,0 +1,321 @@
+/*
+ * 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 java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.CompletableFuture.supplyAsync;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.catalog.commands.CatalogUtils.clusterWideEnsuredActivationTimestamp;
+import static org.apache.ignite.internal.index.IndexManagementUtils.AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC;
+import static org.apache.ignite.internal.index.IndexManagementUtils.earliestCatalogVersionOfIndexInRegisteredStatus;
+import static org.apache.ignite.internal.index.IndexManagementUtils.isPrimaryReplica;
+import static org.apache.ignite.internal.index.IndexManagementUtils.localNode;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeoutException;
+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.CatalogManager;
+import org.apache.ignite.internal.catalog.ClockWaiter;
+import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
+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.index.message.IndexMessagesFactory;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeRequest;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeResponse;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+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.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.PrimaryReplicaAwaitTimeoutException;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.RecipientLeftException;
+
+/**
+ * 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.
+ *
+ * <br><p>Approximate algorithm for the task:</p>
+ * <ul>
+ * <li>Waiting for the activation 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>
+ * <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>
+ * </ul></li>
+ * <li>Change the index status to {@link CatalogIndexStatus#BUILDING}.</li>
+ * </ul>
+ */
+class IndexBuildingStarterTask {
+ private static final IgniteLogger LOG = Loggers.forClass(IndexBuildingStarterTask.class);
+
+ private static final IndexMessagesFactory FACTORY = new IndexMessagesFactory();
+
+ private static final int INVOKE_MESSAGE_TIMEOUT_MILLS = 1_000;
+
+ private static final int RETRY_SEND_MESSAGE_TIMEOUT_MILLS = 100;
+
+ private final int indexId;
+
+ private final int tableId;
+
+ private final CatalogManager catalogManager;
+
+ private final PlacementDriver placementDriver;
+
+ private final ClusterService clusterService;
+
+ private final LogicalTopologyService logicalTopologyService;
+
+ private final HybridClock clock;
+
+ private final ClockWaiter clockWaiter;
+
+ private final Executor executor;
+
+ private final IgniteSpinBusyLock busyLock;
+
+ private volatile int catalogVersionOfIndexCreation = -1;
+
+ private final IgniteSpinBusyLock taskBusyLock = new IgniteSpinBusyLock();
+
+ private final AtomicBoolean taskStopGuard = new AtomicBoolean();
+
+ IndexBuildingStarterTask(
+ int indexId,
+ int tableId,
+ CatalogManager catalogManager,
+ PlacementDriver placementDriver,
+ ClusterService clusterService,
+ LogicalTopologyService logicalTopologyService,
+ HybridClock clock,
+ ClockWaiter clockWaiter,
+ Executor executor,
+ IgniteSpinBusyLock busyLock
+ ) {
+ this.indexId = indexId;
+ this.tableId = tableId;
+ this.catalogManager = catalogManager;
+ this.placementDriver = placementDriver;
+ this.clusterService = clusterService;
+ this.logicalTopologyService = logicalTopologyService;
+ this.clockWaiter = clockWaiter;
+ this.clock = clock;
+ this.executor = executor;
+ this.busyLock = busyLock;
+ }
+
+ /**
+ * Asynchronously starts the task execution.
+ *
+ * @return Future of task execution.
+ */
+ CompletableFuture<Void> start() {
+ if (!enterBusy()) {
+ return nullCompletedFuture();
+ }
+
+ try {
+ return supplyAsync(() -> {
+ int catalogVersionOfIndexCreation = earliestCatalogVersionOfIndexInRegisteredStatus(catalogManager, indexId);
+
+ assert catalogVersionOfIndexCreation >= 0 : indexId;
+
+ this.catalogVersionOfIndexCreation = catalogVersionOfIndexCreation;
+
+ return awaitActivateForCatalogVersionOfIndexCreation()
+ .thenCompose(unused -> ensureThatLocalNodeStillPrimaryReplica())
+ .thenCompose(unused -> inBusyLock(logicalTopologyService::logicalTopologyOnLeader))
+ .thenComposeAsync(this::awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation, executor)
+ .thenComposeAsync(unused -> switchIndexToBuildingStatus(), executor);
+ }, executor)
+ .thenCompose(Function.identity())
+ .handle((unused, throwable) -> {
+ if (throwable != null) {
+ Throwable cause = unwrapCause(throwable);
+
+ if (!(cause instanceof TakStoppingException) && !(cause instanceof NodeStoppingException)) {
Review Comment:
Tak!
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterController.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * 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 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 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.CatalogService;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
+import org.apache.ignite.internal.catalog.events.CatalogEvent;
+import org.apache.ignite.internal.catalog.events.CreateIndexEventParameters;
+import org.apache.ignite.internal.catalog.events.DropIndexEventParameters;
+import org.apache.ignite.internal.close.ManuallyCloseable;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * Component is responsible for starting and stopping {@link IndexBuildingStarterTask} via {@link IndexBuildingStarter}.
+ *
+ * <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>
+ *
+ * <br><p>Tasks are started and stopped based on events:</p>
+ * <ul>
+ * <li>{@link CatalogEvent#INDEX_CREATE} - task starts for the new index.</li>
+ * <li>{@link CatalogEvent#INDEX_DROP} - 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 it all new indexes, and when the primary replica expires, it stops all tasks for the table.</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>
+ */
+class IndexBuildingStarterController implements ManuallyCloseable {
+ private final CatalogService catalogService;
+
+ private final PlacementDriver placementDriver;
+
+ private final ClusterService clusterService;
+
+ private final IndexBuildingStarter indexBuildingStarter;
+
+ /** Tables IDs for which the local node is the primary replica for the partition with ID {@code 0}. */
+ private final Set<Integer> tableIds = ConcurrentHashMap.newKeySet();
Review Comment:
It would be great to have the information about which table IDs these are in the name of the field, like `awaitCoordinatedTableIds`
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterTask.java:
##########
@@ -0,0 +1,321 @@
+/*
+ * 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 java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.CompletableFuture.supplyAsync;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.catalog.commands.CatalogUtils.clusterWideEnsuredActivationTimestamp;
+import static org.apache.ignite.internal.index.IndexManagementUtils.AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC;
+import static org.apache.ignite.internal.index.IndexManagementUtils.earliestCatalogVersionOfIndexInRegisteredStatus;
+import static org.apache.ignite.internal.index.IndexManagementUtils.isPrimaryReplica;
+import static org.apache.ignite.internal.index.IndexManagementUtils.localNode;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeoutException;
+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.CatalogManager;
+import org.apache.ignite.internal.catalog.ClockWaiter;
+import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
+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.index.message.IndexMessagesFactory;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeRequest;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeResponse;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+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.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.PrimaryReplicaAwaitTimeoutException;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.RecipientLeftException;
+
+/**
+ * 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.
+ *
+ * <br><p>Approximate algorithm for the task:</p>
+ * <ul>
+ * <li>Waiting for the activation 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>
+ * <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>
+ * </ul></li>
+ * <li>Change the index status to {@link CatalogIndexStatus#BUILDING}.</li>
+ * </ul>
+ */
+class IndexBuildingStarterTask {
+ private static final IgniteLogger LOG = Loggers.forClass(IndexBuildingStarterTask.class);
+
+ private static final IndexMessagesFactory FACTORY = new IndexMessagesFactory();
+
+ private static final int INVOKE_MESSAGE_TIMEOUT_MILLS = 1_000;
+
+ private static final int RETRY_SEND_MESSAGE_TIMEOUT_MILLS = 100;
+
+ private final int indexId;
+
+ private final int tableId;
+
+ private final CatalogManager catalogManager;
+
+ private final PlacementDriver placementDriver;
+
+ private final ClusterService clusterService;
+
+ private final LogicalTopologyService logicalTopologyService;
+
+ private final HybridClock clock;
+
+ private final ClockWaiter clockWaiter;
+
+ private final Executor executor;
+
+ private final IgniteSpinBusyLock busyLock;
+
+ private volatile int catalogVersionOfIndexCreation = -1;
+
+ private final IgniteSpinBusyLock taskBusyLock = new IgniteSpinBusyLock();
+
+ private final AtomicBoolean taskStopGuard = new AtomicBoolean();
+
+ IndexBuildingStarterTask(
+ int indexId,
+ int tableId,
+ CatalogManager catalogManager,
+ PlacementDriver placementDriver,
+ ClusterService clusterService,
+ LogicalTopologyService logicalTopologyService,
+ HybridClock clock,
+ ClockWaiter clockWaiter,
+ Executor executor,
+ IgniteSpinBusyLock busyLock
+ ) {
+ this.indexId = indexId;
+ this.tableId = tableId;
+ this.catalogManager = catalogManager;
+ this.placementDriver = placementDriver;
+ this.clusterService = clusterService;
+ this.logicalTopologyService = logicalTopologyService;
+ this.clockWaiter = clockWaiter;
+ this.clock = clock;
+ this.executor = executor;
+ this.busyLock = busyLock;
+ }
+
+ /**
+ * Asynchronously starts the task execution.
+ *
+ * @return Future of task execution.
+ */
+ CompletableFuture<Void> start() {
+ if (!enterBusy()) {
+ return nullCompletedFuture();
+ }
+
+ try {
+ return supplyAsync(() -> {
+ int catalogVersionOfIndexCreation = earliestCatalogVersionOfIndexInRegisteredStatus(catalogManager, indexId);
+
+ assert catalogVersionOfIndexCreation >= 0 : indexId;
+
+ this.catalogVersionOfIndexCreation = catalogVersionOfIndexCreation;
+
+ return awaitActivateForCatalogVersionOfIndexCreation()
+ .thenCompose(unused -> ensureThatLocalNodeStillPrimaryReplica())
+ .thenCompose(unused -> inBusyLock(logicalTopologyService::logicalTopologyOnLeader))
+ .thenComposeAsync(this::awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation, executor)
+ .thenComposeAsync(unused -> switchIndexToBuildingStatus(), executor);
+ }, executor)
+ .thenCompose(Function.identity())
+ .handle((unused, throwable) -> {
+ if (throwable != null) {
+ Throwable cause = unwrapCause(throwable);
+
+ if (!(cause instanceof TakStoppingException) && !(cause instanceof NodeStoppingException)) {
+ LOG.error("Error starting index building: ", cause, indexId);
+ }
+ }
+
+ return null;
+ });
+ } finally {
+ leaveBusy();
+ }
+ }
+
+ /** Stops the execution of a task. */
+ void stop() {
+ if (!taskStopGuard.compareAndSet(false, true)) {
+ return;
+ }
+
+ taskBusyLock.block();
+ }
+
+ private CompletableFuture<Void> awaitActivateForCatalogVersionOfIndexCreation() {
+ return inBusyLock(() -> {
+ Catalog catalog = catalogManager.catalog(catalogVersionOfIndexCreation);
+
+ assert catalog != null : IgniteStringFormatter.format("Missing catalog version: [indexId={}, catalogVersion={}]",
+ indexId, catalogVersionOfIndexCreation);
+
+ return clockWaiter.waitFor(clusterWideEnsuredActivationTimestamp(catalog));
+ });
+ }
+
+ private CompletableFuture<Void> ensureThatLocalNodeStillPrimaryReplica() {
+ return awaitPrimaryReplica().thenAccept(replicaMeta -> {
+ if (!enterBusy()) {
+ throw new TakStoppingException();
+ }
+
+ try {
+ if (!isPrimaryReplica(replicaMeta, localNode(clusterService), clock.now())) {
+ // Lease has expired, we stop the task.
+ throw new TakStoppingException();
+ }
+ } finally {
+ leaveBusy();
+ }
+ });
+ }
+
+ private CompletableFuture<ReplicaMeta> awaitPrimaryReplica() {
+ return inBusyLock(() -> {
+ TablePartitionId groupId = new TablePartitionId(tableId, 0);
+
+ return placementDriver.awaitPrimaryReplica(groupId, clock.now(), AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC, SECONDS)
+ .handle((replicaMeta, throwable) -> {
+ if (throwable != null) {
+ Throwable cause = unwrapCause(throwable);
+
+ if (cause instanceof PrimaryReplicaAwaitTimeoutException) {
+ return awaitPrimaryReplica();
+ }
+
+ return CompletableFuture.<ReplicaMeta>failedFuture(cause);
+ }
+
+ return completedFuture(replicaMeta);
+ }).thenCompose(Function.identity());
+ });
+ }
+
+ private CompletableFuture<Void> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(LogicalTopologySnapshot logicalTopologySnapshot) {
+ return inBusyLock(() -> {
+ Set<LogicalNode> remainingNodes = ConcurrentHashMap.newKeySet();
+ remainingNodes.addAll(logicalTopologySnapshot.nodes());
+
+ var nodeLeftLogicalTopologyListener = new NodeLeftLogicalTopologyListener(remainingNodes);
+ logicalTopologyService.addEventListener(nodeLeftLogicalTopologyListener);
+
+ CompletableFuture[] futures = remainingNodes.stream()
+ .map(node -> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(node, remainingNodes))
+ .toArray(CompletableFuture[]::new);
+
+ return allOf(futures)
+ .whenComplete((unused, throwable) -> logicalTopologyService.removeEventListener(nodeLeftLogicalTopologyListener));
+ });
+ }
+
+ private CompletableFuture<NetworkMessage> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(
+ LogicalNode targetNode,
+ Set<LogicalNode> remainingNodes
+ ) {
+ return inBusyLock(() -> {
+ if (!remainingNodes.contains(targetNode)) {
+ // Target node left logical topology, there is no need to do anything.
+ return nullCompletedFuture();
+ }
+
+ IsNodeFinishedRwTransactionsStartedBeforeRequest request = isNodeFinishedRwTransactionsStartedBeforeRequest();
+
+ return clusterService.messagingService().invoke(targetNode, request, INVOKE_MESSAGE_TIMEOUT_MILLS)
+ .handle((message, throwable) -> {
+ if (throwable != null) {
+ Throwable cause = unwrapCause(throwable);
+
+ if (!(cause instanceof TimeoutException) && !(cause instanceof RecipientLeftException)) {
+ return CompletableFuture.<NetworkMessage>failedFuture(cause);
+ }
+ } else if (((IsNodeFinishedRwTransactionsStartedBeforeResponse) message).finished()) {
+ remainingNodes.remove(targetNode);
+
+ return completedFuture(message);
+ }
+
+ // Let's try to send the message again, but after a short time.
+ return inBusyLock(() -> clockWaiter.waitFor(clock.now().addPhysicalTime(RETRY_SEND_MESSAGE_TIMEOUT_MILLS)))
+ .thenCompose(unused -> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(targetNode, remainingNodes));
+ }).thenCompose(Function.identity());
+ });
+ }
+
+ private CompletableFuture<Void> switchIndexToBuildingStatus() {
+ return inBusyLock(() -> catalogManager.execute(StartBuildingIndexCommand.builder().indexId(indexId).build()));
+ }
+
+ private IsNodeFinishedRwTransactionsStartedBeforeRequest isNodeFinishedRwTransactionsStartedBeforeRequest() {
+ return FACTORY.isNodeFinishedRwTransactionsStartedBeforeRequest().targetCatalogVersion(catalogVersionOfIndexCreation).build();
+ }
+
+ private boolean enterBusy() {
+ return IndexManagementUtils.enterBusy(busyLock, taskBusyLock);
+ }
+
+ private void leaveBusy() {
+ IndexManagementUtils.leaveBusy(busyLock, taskBusyLock);
+ }
+
+ private <T> CompletableFuture<T> inBusyLock(Supplier<CompletableFuture<T>> supplier) {
Review Comment:
```suggestion
private <T> CompletableFuture<T> inBusyLocks(Supplier<CompletableFuture<T>> supplier) {
```
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterTask.java:
##########
@@ -0,0 +1,321 @@
+/*
+ * 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 java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.CompletableFuture.supplyAsync;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.catalog.commands.CatalogUtils.clusterWideEnsuredActivationTimestamp;
+import static org.apache.ignite.internal.index.IndexManagementUtils.AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC;
+import static org.apache.ignite.internal.index.IndexManagementUtils.earliestCatalogVersionOfIndexInRegisteredStatus;
+import static org.apache.ignite.internal.index.IndexManagementUtils.isPrimaryReplica;
+import static org.apache.ignite.internal.index.IndexManagementUtils.localNode;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeoutException;
+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.CatalogManager;
+import org.apache.ignite.internal.catalog.ClockWaiter;
+import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
+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.index.message.IndexMessagesFactory;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeRequest;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeResponse;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+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.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.PrimaryReplicaAwaitTimeoutException;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.RecipientLeftException;
+
+/**
+ * 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.
+ *
+ * <br><p>Approximate algorithm for the task:</p>
+ * <ul>
+ * <li>Waiting for the activation 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>
+ * <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>
+ * </ul></li>
+ * <li>Change the index status to {@link CatalogIndexStatus#BUILDING}.</li>
+ * </ul>
+ */
+class IndexBuildingStarterTask {
+ private static final IgniteLogger LOG = Loggers.forClass(IndexBuildingStarterTask.class);
+
+ private static final IndexMessagesFactory FACTORY = new IndexMessagesFactory();
+
+ private static final int INVOKE_MESSAGE_TIMEOUT_MILLS = 1_000;
+
+ private static final int RETRY_SEND_MESSAGE_TIMEOUT_MILLS = 100;
+
+ private final int indexId;
+
+ private final int tableId;
+
+ private final CatalogManager catalogManager;
+
+ private final PlacementDriver placementDriver;
+
+ private final ClusterService clusterService;
+
+ private final LogicalTopologyService logicalTopologyService;
+
+ private final HybridClock clock;
+
+ private final ClockWaiter clockWaiter;
+
+ private final Executor executor;
+
+ private final IgniteSpinBusyLock busyLock;
+
+ private volatile int catalogVersionOfIndexCreation = -1;
+
+ private final IgniteSpinBusyLock taskBusyLock = new IgniteSpinBusyLock();
+
+ private final AtomicBoolean taskStopGuard = new AtomicBoolean();
+
+ IndexBuildingStarterTask(
+ int indexId,
+ int tableId,
+ CatalogManager catalogManager,
+ PlacementDriver placementDriver,
+ ClusterService clusterService,
+ LogicalTopologyService logicalTopologyService,
+ HybridClock clock,
+ ClockWaiter clockWaiter,
+ Executor executor,
+ IgniteSpinBusyLock busyLock
+ ) {
+ this.indexId = indexId;
+ this.tableId = tableId;
+ this.catalogManager = catalogManager;
+ this.placementDriver = placementDriver;
+ this.clusterService = clusterService;
+ this.logicalTopologyService = logicalTopologyService;
+ this.clockWaiter = clockWaiter;
+ this.clock = clock;
+ this.executor = executor;
+ this.busyLock = busyLock;
+ }
+
+ /**
+ * Asynchronously starts the task execution.
+ *
+ * @return Future of task execution.
+ */
+ CompletableFuture<Void> start() {
+ if (!enterBusy()) {
+ return nullCompletedFuture();
+ }
+
+ try {
+ return supplyAsync(() -> {
+ int catalogVersionOfIndexCreation = earliestCatalogVersionOfIndexInRegisteredStatus(catalogManager, indexId);
+
+ assert catalogVersionOfIndexCreation >= 0 : indexId;
+
+ this.catalogVersionOfIndexCreation = catalogVersionOfIndexCreation;
+
+ return awaitActivateForCatalogVersionOfIndexCreation()
+ .thenCompose(unused -> ensureThatLocalNodeStillPrimaryReplica())
+ .thenCompose(unused -> inBusyLock(logicalTopologyService::logicalTopologyOnLeader))
+ .thenComposeAsync(this::awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation, executor)
+ .thenComposeAsync(unused -> switchIndexToBuildingStatus(), executor);
+ }, executor)
+ .thenCompose(Function.identity())
+ .handle((unused, throwable) -> {
+ if (throwable != null) {
+ Throwable cause = unwrapCause(throwable);
+
+ if (!(cause instanceof TakStoppingException) && !(cause instanceof NodeStoppingException)) {
+ LOG.error("Error starting index building: ", cause, indexId);
+ }
+ }
+
+ return null;
+ });
+ } finally {
+ leaveBusy();
+ }
+ }
+
+ /** Stops the execution of a task. */
+ void stop() {
+ if (!taskStopGuard.compareAndSet(false, true)) {
+ return;
+ }
+
+ taskBusyLock.block();
+ }
+
+ private CompletableFuture<Void> awaitActivateForCatalogVersionOfIndexCreation() {
+ return inBusyLock(() -> {
+ Catalog catalog = catalogManager.catalog(catalogVersionOfIndexCreation);
+
+ assert catalog != null : IgniteStringFormatter.format("Missing catalog version: [indexId={}, catalogVersion={}]",
+ indexId, catalogVersionOfIndexCreation);
+
+ return clockWaiter.waitFor(clusterWideEnsuredActivationTimestamp(catalog));
+ });
+ }
+
+ private CompletableFuture<Void> ensureThatLocalNodeStillPrimaryReplica() {
+ return awaitPrimaryReplica().thenAccept(replicaMeta -> {
+ if (!enterBusy()) {
+ throw new TakStoppingException();
+ }
+
+ try {
+ if (!isPrimaryReplica(replicaMeta, localNode(clusterService), clock.now())) {
+ // Lease has expired, we stop the task.
+ throw new TakStoppingException();
+ }
+ } finally {
+ leaveBusy();
+ }
+ });
+ }
+
+ private CompletableFuture<ReplicaMeta> awaitPrimaryReplica() {
+ return inBusyLock(() -> {
+ TablePartitionId groupId = new TablePartitionId(tableId, 0);
+
+ return placementDriver.awaitPrimaryReplica(groupId, clock.now(), AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC, SECONDS)
+ .handle((replicaMeta, throwable) -> {
+ if (throwable != null) {
+ Throwable cause = unwrapCause(throwable);
+
+ if (cause instanceof PrimaryReplicaAwaitTimeoutException) {
+ return awaitPrimaryReplica();
+ }
+
+ return CompletableFuture.<ReplicaMeta>failedFuture(cause);
+ }
+
+ return completedFuture(replicaMeta);
+ }).thenCompose(Function.identity());
+ });
+ }
+
+ private CompletableFuture<Void> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(LogicalTopologySnapshot logicalTopologySnapshot) {
+ return inBusyLock(() -> {
+ Set<LogicalNode> remainingNodes = ConcurrentHashMap.newKeySet();
+ remainingNodes.addAll(logicalTopologySnapshot.nodes());
+
+ var nodeLeftLogicalTopologyListener = new NodeLeftLogicalTopologyListener(remainingNodes);
+ logicalTopologyService.addEventListener(nodeLeftLogicalTopologyListener);
+
+ CompletableFuture[] futures = remainingNodes.stream()
+ .map(node -> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(node, remainingNodes))
+ .toArray(CompletableFuture[]::new);
+
+ return allOf(futures)
+ .whenComplete((unused, throwable) -> logicalTopologyService.removeEventListener(nodeLeftLogicalTopologyListener));
+ });
+ }
+
+ private CompletableFuture<NetworkMessage> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(
+ LogicalNode targetNode,
+ Set<LogicalNode> remainingNodes
+ ) {
+ return inBusyLock(() -> {
+ if (!remainingNodes.contains(targetNode)) {
+ // Target node left logical topology, there is no need to do anything.
+ return nullCompletedFuture();
+ }
+
+ IsNodeFinishedRwTransactionsStartedBeforeRequest request = isNodeFinishedRwTransactionsStartedBeforeRequest();
+
+ return clusterService.messagingService().invoke(targetNode, request, INVOKE_MESSAGE_TIMEOUT_MILLS)
+ .handle((message, throwable) -> {
+ if (throwable != null) {
+ Throwable cause = unwrapCause(throwable);
+
+ if (!(cause instanceof TimeoutException) && !(cause instanceof RecipientLeftException)) {
+ return CompletableFuture.<NetworkMessage>failedFuture(cause);
+ }
+ } else if (((IsNodeFinishedRwTransactionsStartedBeforeResponse) message).finished()) {
+ remainingNodes.remove(targetNode);
+
+ return completedFuture(message);
+ }
+
+ // Let's try to send the message again, but after a short time.
+ return inBusyLock(() -> clockWaiter.waitFor(clock.now().addPhysicalTime(RETRY_SEND_MESSAGE_TIMEOUT_MILLS)))
Review Comment:
I'm not sure we should use Clock and ClockWaiter just to schedule a task after X milliseconds. For this, a usual `ScheduledThreadPool` should be enough. There is a common scheduled pool in `ThreadPoolsManager`, for example.
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterTask.java:
##########
@@ -0,0 +1,321 @@
+/*
+ * 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 java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.CompletableFuture.supplyAsync;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.catalog.commands.CatalogUtils.clusterWideEnsuredActivationTimestamp;
+import static org.apache.ignite.internal.index.IndexManagementUtils.AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC;
+import static org.apache.ignite.internal.index.IndexManagementUtils.earliestCatalogVersionOfIndexInRegisteredStatus;
+import static org.apache.ignite.internal.index.IndexManagementUtils.isPrimaryReplica;
+import static org.apache.ignite.internal.index.IndexManagementUtils.localNode;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeoutException;
+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.CatalogManager;
+import org.apache.ignite.internal.catalog.ClockWaiter;
+import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
+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.index.message.IndexMessagesFactory;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeRequest;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeResponse;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+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.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.PrimaryReplicaAwaitTimeoutException;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.RecipientLeftException;
+
+/**
+ * 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.
+ *
+ * <br><p>Approximate algorithm for the task:</p>
+ * <ul>
+ * <li>Waiting for the activation of the catalog version in which the index was created.</li>
Review Comment:
```suggestion
* <li>Waiting for the activation (on every cluster node) of the catalog version in which the index was created.</li>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-21115 Wait for completion of transactions started before index appearance [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #3095:
URL: https://github.com/apache/ignite-3/pull/3095#discussion_r1470637942
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterTask.java:
##########
@@ -0,0 +1,321 @@
+/*
+ * 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 java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.CompletableFuture.supplyAsync;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.catalog.commands.CatalogUtils.clusterWideEnsuredActivationTimestamp;
+import static org.apache.ignite.internal.index.IndexManagementUtils.AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC;
+import static org.apache.ignite.internal.index.IndexManagementUtils.earliestCatalogVersionOfIndexInRegisteredStatus;
+import static org.apache.ignite.internal.index.IndexManagementUtils.isPrimaryReplica;
+import static org.apache.ignite.internal.index.IndexManagementUtils.localNode;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeoutException;
+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.CatalogManager;
+import org.apache.ignite.internal.catalog.ClockWaiter;
+import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
+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.index.message.IndexMessagesFactory;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeRequest;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeResponse;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+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.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.PrimaryReplicaAwaitTimeoutException;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.RecipientLeftException;
+
+/**
+ * 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.
+ *
+ * <br><p>Approximate algorithm for the task:</p>
+ * <ul>
+ * <li>Waiting for the activation 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>
+ * <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>
+ * </ul></li>
+ * <li>Change the index status to {@link CatalogIndexStatus#BUILDING}.</li>
+ * </ul>
+ */
+class IndexBuildingStarterTask {
+ private static final IgniteLogger LOG = Loggers.forClass(IndexBuildingStarterTask.class);
+
+ private static final IndexMessagesFactory FACTORY = new IndexMessagesFactory();
+
+ private static final int INVOKE_MESSAGE_TIMEOUT_MILLS = 1_000;
+
+ private static final int RETRY_SEND_MESSAGE_TIMEOUT_MILLS = 100;
+
+ private final int indexId;
+
+ private final int tableId;
+
+ private final CatalogManager catalogManager;
+
+ private final PlacementDriver placementDriver;
+
+ private final ClusterService clusterService;
+
+ private final LogicalTopologyService logicalTopologyService;
+
+ private final HybridClock clock;
+
+ private final ClockWaiter clockWaiter;
+
+ private final Executor executor;
+
+ private final IgniteSpinBusyLock busyLock;
+
+ private volatile int catalogVersionOfIndexCreation = -1;
+
+ private final IgniteSpinBusyLock taskBusyLock = new IgniteSpinBusyLock();
+
+ private final AtomicBoolean taskStopGuard = new AtomicBoolean();
+
+ IndexBuildingStarterTask(
+ int indexId,
+ int tableId,
+ CatalogManager catalogManager,
+ PlacementDriver placementDriver,
+ ClusterService clusterService,
+ LogicalTopologyService logicalTopologyService,
+ HybridClock clock,
+ ClockWaiter clockWaiter,
+ Executor executor,
+ IgniteSpinBusyLock busyLock
+ ) {
+ this.indexId = indexId;
+ this.tableId = tableId;
+ this.catalogManager = catalogManager;
+ this.placementDriver = placementDriver;
+ this.clusterService = clusterService;
+ this.logicalTopologyService = logicalTopologyService;
+ this.clockWaiter = clockWaiter;
+ this.clock = clock;
+ this.executor = executor;
+ this.busyLock = busyLock;
+ }
+
+ /**
+ * Asynchronously starts the task execution.
+ *
+ * @return Future of task execution.
+ */
+ CompletableFuture<Void> start() {
+ if (!enterBusy()) {
+ return nullCompletedFuture();
+ }
+
+ try {
+ return supplyAsync(() -> {
+ int catalogVersionOfIndexCreation = earliestCatalogVersionOfIndexInRegisteredStatus(catalogManager, indexId);
+
+ assert catalogVersionOfIndexCreation >= 0 : indexId;
+
+ this.catalogVersionOfIndexCreation = catalogVersionOfIndexCreation;
+
+ return awaitActivateForCatalogVersionOfIndexCreation()
+ .thenCompose(unused -> ensureThatLocalNodeStillPrimaryReplica())
+ .thenCompose(unused -> inBusyLock(logicalTopologyService::logicalTopologyOnLeader))
+ .thenComposeAsync(this::awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation, executor)
+ .thenComposeAsync(unused -> switchIndexToBuildingStatus(), executor);
+ }, executor)
+ .thenCompose(Function.identity())
+ .handle((unused, throwable) -> {
+ if (throwable != null) {
+ Throwable cause = unwrapCause(throwable);
+
+ if (!(cause instanceof TakStoppingException) && !(cause instanceof NodeStoppingException)) {
Review Comment:
Rename it =)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-21115 Wait for completion of transactions started before index appearance [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill merged PR #3095:
URL: https://github.com/apache/ignite-3/pull/3095
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-21115 Wait for completion of transactions started before index appearance [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #3095:
URL: https://github.com/apache/ignite-3/pull/3095#discussion_r1470652077
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterController.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * 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 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 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.CatalogService;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
+import org.apache.ignite.internal.catalog.events.CatalogEvent;
+import org.apache.ignite.internal.catalog.events.CreateIndexEventParameters;
+import org.apache.ignite.internal.catalog.events.DropIndexEventParameters;
+import org.apache.ignite.internal.close.ManuallyCloseable;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * Component is responsible for starting and stopping {@link IndexBuildingStarterTask} via {@link IndexBuildingStarter}.
+ *
+ * <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>
+ *
+ * <br><p>Tasks are started and stopped based on events:</p>
+ * <ul>
+ * <li>{@link CatalogEvent#INDEX_CREATE} - task starts for the new index.</li>
+ * <li>{@link CatalogEvent#INDEX_DROP} - 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 it all new indexes, and when the primary replica expires, it stops all tasks for the table.</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>
+ */
+class IndexBuildingStarterController implements ManuallyCloseable {
+ private final CatalogService catalogService;
+
+ private final PlacementDriver placementDriver;
+
+ private final ClusterService clusterService;
+
+ private final IndexBuildingStarter indexBuildingStarter;
+
+ /** Tables IDs for which the local node is the primary replica for the partition with ID {@code 0}. */
+ private final Set<Integer> tableIds = ConcurrentHashMap.newKeySet();
Review Comment:
I tried renaming it differently.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-21115 Wait for completion of transactions started before index appearance [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #3095:
URL: https://github.com/apache/ignite-3/pull/3095#discussion_r1470674677
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingManager.java:
##########
@@ -79,16 +80,45 @@ public IndexBuildingManager(
IndexManager indexManager,
PlacementDriver placementDriver,
ClusterService clusterService,
- HybridClock clock
+ LogicalTopologyService logicalTopologyService,
+ HybridClock clock,
+ ClockWaiter clockWaiter
) {
this.metaStorageManager = metaStorageManager;
- this.catalogManager = catalogManager;
- indexBuilder = new IndexBuilder(nodeName, Runtime.getRuntime().availableProcessors(), replicaService);
+ int threadCount = Runtime.getRuntime().availableProcessors();
+
+ executor = new ThreadPoolExecutor(
+ threadCount,
Review Comment:
Try fix it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-21115 Wait for completion of transactions started before index appearance [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #3095:
URL: https://github.com/apache/ignite-3/pull/3095#discussion_r1470631531
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterTask.java:
##########
@@ -0,0 +1,321 @@
+/*
+ * 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 java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.CompletableFuture.supplyAsync;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.catalog.commands.CatalogUtils.clusterWideEnsuredActivationTimestamp;
+import static org.apache.ignite.internal.index.IndexManagementUtils.AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC;
+import static org.apache.ignite.internal.index.IndexManagementUtils.earliestCatalogVersionOfIndexInRegisteredStatus;
+import static org.apache.ignite.internal.index.IndexManagementUtils.isPrimaryReplica;
+import static org.apache.ignite.internal.index.IndexManagementUtils.localNode;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeoutException;
+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.CatalogManager;
+import org.apache.ignite.internal.catalog.ClockWaiter;
+import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
+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.index.message.IndexMessagesFactory;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeRequest;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeResponse;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+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.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.PrimaryReplicaAwaitTimeoutException;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.RecipientLeftException;
+
+/**
+ * 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.
+ *
+ * <br><p>Approximate algorithm for the task:</p>
+ * <ul>
+ * <li>Waiting for the activation 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>
+ * <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>
+ * </ul></li>
+ * <li>Change the index status to {@link CatalogIndexStatus#BUILDING}.</li>
+ * </ul>
+ */
+class IndexBuildingStarterTask {
+ private static final IgniteLogger LOG = Loggers.forClass(IndexBuildingStarterTask.class);
+
+ private static final IndexMessagesFactory FACTORY = new IndexMessagesFactory();
+
+ private static final int INVOKE_MESSAGE_TIMEOUT_MILLS = 1_000;
+
+ private static final int RETRY_SEND_MESSAGE_TIMEOUT_MILLS = 100;
+
+ private final int indexId;
+
+ private final int tableId;
+
+ private final CatalogManager catalogManager;
+
+ private final PlacementDriver placementDriver;
+
+ private final ClusterService clusterService;
+
+ private final LogicalTopologyService logicalTopologyService;
+
+ private final HybridClock clock;
+
+ private final ClockWaiter clockWaiter;
+
+ private final Executor executor;
+
+ private final IgniteSpinBusyLock busyLock;
+
+ private volatile int catalogVersionOfIndexCreation = -1;
+
+ private final IgniteSpinBusyLock taskBusyLock = new IgniteSpinBusyLock();
+
+ private final AtomicBoolean taskStopGuard = new AtomicBoolean();
+
+ IndexBuildingStarterTask(
+ int indexId,
+ int tableId,
+ CatalogManager catalogManager,
+ PlacementDriver placementDriver,
+ ClusterService clusterService,
+ LogicalTopologyService logicalTopologyService,
+ HybridClock clock,
+ ClockWaiter clockWaiter,
+ Executor executor,
+ IgniteSpinBusyLock busyLock
+ ) {
+ this.indexId = indexId;
+ this.tableId = tableId;
+ this.catalogManager = catalogManager;
+ this.placementDriver = placementDriver;
+ this.clusterService = clusterService;
+ this.logicalTopologyService = logicalTopologyService;
+ this.clockWaiter = clockWaiter;
+ this.clock = clock;
+ this.executor = executor;
+ this.busyLock = busyLock;
+ }
+
+ /**
+ * Asynchronously starts the task execution.
+ *
+ * @return Future of task execution.
+ */
+ CompletableFuture<Void> start() {
+ if (!enterBusy()) {
+ return nullCompletedFuture();
+ }
+
+ try {
+ return supplyAsync(() -> {
+ int catalogVersionOfIndexCreation = earliestCatalogVersionOfIndexInRegisteredStatus(catalogManager, indexId);
+
+ assert catalogVersionOfIndexCreation >= 0 : indexId;
+
+ this.catalogVersionOfIndexCreation = catalogVersionOfIndexCreation;
+
+ return awaitActivateForCatalogVersionOfIndexCreation()
+ .thenCompose(unused -> ensureThatLocalNodeStillPrimaryReplica())
+ .thenCompose(unused -> inBusyLock(logicalTopologyService::logicalTopologyOnLeader))
+ .thenComposeAsync(this::awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation, executor)
+ .thenComposeAsync(unused -> switchIndexToBuildingStatus(), executor);
+ }, executor)
+ .thenCompose(Function.identity())
+ .handle((unused, throwable) -> {
+ if (throwable != null) {
+ Throwable cause = unwrapCause(throwable);
+
+ if (!(cause instanceof TakStoppingException) && !(cause instanceof NodeStoppingException)) {
+ LOG.error("Error starting index building: ", cause, indexId);
+ }
+ }
+
+ return null;
+ });
+ } finally {
+ leaveBusy();
+ }
+ }
+
+ /** Stops the execution of a task. */
+ void stop() {
+ if (!taskStopGuard.compareAndSet(false, true)) {
+ return;
+ }
+
+ taskBusyLock.block();
+ }
+
+ private CompletableFuture<Void> awaitActivateForCatalogVersionOfIndexCreation() {
+ return inBusyLock(() -> {
+ Catalog catalog = catalogManager.catalog(catalogVersionOfIndexCreation);
+
+ assert catalog != null : IgniteStringFormatter.format("Missing catalog version: [indexId={}, catalogVersion={}]",
+ indexId, catalogVersionOfIndexCreation);
+
+ return clockWaiter.waitFor(clusterWideEnsuredActivationTimestamp(catalog));
+ });
+ }
+
+ private CompletableFuture<Void> ensureThatLocalNodeStillPrimaryReplica() {
+ return awaitPrimaryReplica().thenAccept(replicaMeta -> {
+ if (!enterBusy()) {
+ throw new TakStoppingException();
+ }
+
+ try {
+ if (!isPrimaryReplica(replicaMeta, localNode(clusterService), clock.now())) {
+ // Lease has expired, we stop the task.
+ throw new TakStoppingException();
+ }
+ } finally {
+ leaveBusy();
+ }
+ });
+ }
+
+ private CompletableFuture<ReplicaMeta> awaitPrimaryReplica() {
+ return inBusyLock(() -> {
+ TablePartitionId groupId = new TablePartitionId(tableId, 0);
+
+ return placementDriver.awaitPrimaryReplica(groupId, clock.now(), AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC, SECONDS)
+ .handle((replicaMeta, throwable) -> {
+ if (throwable != null) {
+ Throwable cause = unwrapCause(throwable);
+
+ if (cause instanceof PrimaryReplicaAwaitTimeoutException) {
+ return awaitPrimaryReplica();
+ }
+
+ return CompletableFuture.<ReplicaMeta>failedFuture(cause);
+ }
+
+ return completedFuture(replicaMeta);
+ }).thenCompose(Function.identity());
+ });
+ }
+
+ private CompletableFuture<Void> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(LogicalTopologySnapshot logicalTopologySnapshot) {
+ return inBusyLock(() -> {
+ Set<LogicalNode> remainingNodes = ConcurrentHashMap.newKeySet();
+ remainingNodes.addAll(logicalTopologySnapshot.nodes());
+
+ var nodeLeftLogicalTopologyListener = new NodeLeftLogicalTopologyListener(remainingNodes);
+ logicalTopologyService.addEventListener(nodeLeftLogicalTopologyListener);
+
+ CompletableFuture[] futures = remainingNodes.stream()
+ .map(node -> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(node, remainingNodes))
+ .toArray(CompletableFuture[]::new);
+
+ return allOf(futures)
+ .whenComplete((unused, throwable) -> logicalTopologyService.removeEventListener(nodeLeftLogicalTopologyListener));
+ });
+ }
+
+ private CompletableFuture<NetworkMessage> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(
+ LogicalNode targetNode,
+ Set<LogicalNode> remainingNodes
+ ) {
+ return inBusyLock(() -> {
+ if (!remainingNodes.contains(targetNode)) {
+ // Target node left logical topology, there is no need to do anything.
+ return nullCompletedFuture();
+ }
+
+ IsNodeFinishedRwTransactionsStartedBeforeRequest request = isNodeFinishedRwTransactionsStartedBeforeRequest();
+
+ return clusterService.messagingService().invoke(targetNode, request, INVOKE_MESSAGE_TIMEOUT_MILLS)
+ .handle((message, throwable) -> {
+ if (throwable != null) {
+ Throwable cause = unwrapCause(throwable);
+
+ if (!(cause instanceof TimeoutException) && !(cause instanceof RecipientLeftException)) {
+ return CompletableFuture.<NetworkMessage>failedFuture(cause);
+ }
+ } else if (((IsNodeFinishedRwTransactionsStartedBeforeResponse) message).finished()) {
+ remainingNodes.remove(targetNode);
+
+ return completedFuture(message);
+ }
+
+ // Let's try to send the message again, but after a short time.
+ return inBusyLock(() -> clockWaiter.waitFor(clock.now().addPhysicalTime(RETRY_SEND_MESSAGE_TIMEOUT_MILLS)))
+ .thenCompose(unused -> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(targetNode, remainingNodes));
+ }).thenCompose(Function.identity());
+ });
+ }
+
+ private CompletableFuture<Void> switchIndexToBuildingStatus() {
+ return inBusyLock(() -> catalogManager.execute(StartBuildingIndexCommand.builder().indexId(indexId).build()));
+ }
+
+ private IsNodeFinishedRwTransactionsStartedBeforeRequest isNodeFinishedRwTransactionsStartedBeforeRequest() {
+ return FACTORY.isNodeFinishedRwTransactionsStartedBeforeRequest().targetCatalogVersion(catalogVersionOfIndexCreation).build();
+ }
+
+ private boolean enterBusy() {
+ return IndexManagementUtils.enterBusy(busyLock, taskBusyLock);
+ }
+
+ private void leaveBusy() {
+ IndexManagementUtils.leaveBusy(busyLock, taskBusyLock);
+ }
+
+ private <T> CompletableFuture<T> inBusyLock(Supplier<CompletableFuture<T>> supplier) {
Review Comment:
why not
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-21115 Wait for completion of transactions started before index appearance [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #3095:
URL: https://github.com/apache/ignite-3/pull/3095#discussion_r1470629983
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterController.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * 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 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 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.CatalogService;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
+import org.apache.ignite.internal.catalog.events.CatalogEvent;
+import org.apache.ignite.internal.catalog.events.CreateIndexEventParameters;
+import org.apache.ignite.internal.catalog.events.DropIndexEventParameters;
+import org.apache.ignite.internal.close.ManuallyCloseable;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * Component is responsible for starting and stopping {@link IndexBuildingStarterTask} via {@link IndexBuildingStarter}.
+ *
+ * <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>
+ *
+ * <br><p>Tasks are started and stopped based on events:</p>
+ * <ul>
+ * <li>{@link CatalogEvent#INDEX_CREATE} - task starts for the new index.</li>
+ * <li>{@link CatalogEvent#INDEX_DROP} - 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 it all new indexes, and when the primary replica expires, it stops all tasks for the table.</li>
Review Comment:
apply it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-21115 Wait for completion of transactions started before index appearance [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #3095:
URL: https://github.com/apache/ignite-3/pull/3095#discussion_r1470721773
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterTask.java:
##########
@@ -0,0 +1,321 @@
+/*
+ * 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 java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.CompletableFuture.supplyAsync;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.catalog.commands.CatalogUtils.clusterWideEnsuredActivationTimestamp;
+import static org.apache.ignite.internal.index.IndexManagementUtils.AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC;
+import static org.apache.ignite.internal.index.IndexManagementUtils.earliestCatalogVersionOfIndexInRegisteredStatus;
+import static org.apache.ignite.internal.index.IndexManagementUtils.isPrimaryReplica;
+import static org.apache.ignite.internal.index.IndexManagementUtils.localNode;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeoutException;
+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.CatalogManager;
+import org.apache.ignite.internal.catalog.ClockWaiter;
+import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
+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.index.message.IndexMessagesFactory;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeRequest;
+import org.apache.ignite.internal.index.message.IsNodeFinishedRwTransactionsStartedBeforeResponse;
+import org.apache.ignite.internal.lang.IgniteStringFormatter;
+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.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.PrimaryReplicaAwaitTimeoutException;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.RecipientLeftException;
+
+/**
+ * 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.
+ *
+ * <br><p>Approximate algorithm for the task:</p>
+ * <ul>
+ * <li>Waiting for the activation 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>
+ * <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>
+ * </ul></li>
+ * <li>Change the index status to {@link CatalogIndexStatus#BUILDING}.</li>
+ * </ul>
+ */
+class IndexBuildingStarterTask {
+ private static final IgniteLogger LOG = Loggers.forClass(IndexBuildingStarterTask.class);
+
+ private static final IndexMessagesFactory FACTORY = new IndexMessagesFactory();
+
+ private static final int INVOKE_MESSAGE_TIMEOUT_MILLS = 1_000;
+
+ private static final int RETRY_SEND_MESSAGE_TIMEOUT_MILLS = 100;
+
+ private final int indexId;
+
+ private final int tableId;
+
+ private final CatalogManager catalogManager;
+
+ private final PlacementDriver placementDriver;
+
+ private final ClusterService clusterService;
+
+ private final LogicalTopologyService logicalTopologyService;
+
+ private final HybridClock clock;
+
+ private final ClockWaiter clockWaiter;
+
+ private final Executor executor;
+
+ private final IgniteSpinBusyLock busyLock;
+
+ private volatile int catalogVersionOfIndexCreation = -1;
+
+ private final IgniteSpinBusyLock taskBusyLock = new IgniteSpinBusyLock();
+
+ private final AtomicBoolean taskStopGuard = new AtomicBoolean();
+
+ IndexBuildingStarterTask(
+ int indexId,
+ int tableId,
+ CatalogManager catalogManager,
+ PlacementDriver placementDriver,
+ ClusterService clusterService,
+ LogicalTopologyService logicalTopologyService,
+ HybridClock clock,
+ ClockWaiter clockWaiter,
+ Executor executor,
+ IgniteSpinBusyLock busyLock
+ ) {
+ this.indexId = indexId;
+ this.tableId = tableId;
+ this.catalogManager = catalogManager;
+ this.placementDriver = placementDriver;
+ this.clusterService = clusterService;
+ this.logicalTopologyService = logicalTopologyService;
+ this.clockWaiter = clockWaiter;
+ this.clock = clock;
+ this.executor = executor;
+ this.busyLock = busyLock;
+ }
+
+ /**
+ * Asynchronously starts the task execution.
+ *
+ * @return Future of task execution.
+ */
+ CompletableFuture<Void> start() {
+ if (!enterBusy()) {
+ return nullCompletedFuture();
+ }
+
+ try {
+ return supplyAsync(() -> {
+ int catalogVersionOfIndexCreation = earliestCatalogVersionOfIndexInRegisteredStatus(catalogManager, indexId);
+
+ assert catalogVersionOfIndexCreation >= 0 : indexId;
+
+ this.catalogVersionOfIndexCreation = catalogVersionOfIndexCreation;
+
+ return awaitActivateForCatalogVersionOfIndexCreation()
+ .thenCompose(unused -> ensureThatLocalNodeStillPrimaryReplica())
+ .thenCompose(unused -> inBusyLock(logicalTopologyService::logicalTopologyOnLeader))
+ .thenComposeAsync(this::awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation, executor)
+ .thenComposeAsync(unused -> switchIndexToBuildingStatus(), executor);
+ }, executor)
+ .thenCompose(Function.identity())
+ .handle((unused, throwable) -> {
+ if (throwable != null) {
+ Throwable cause = unwrapCause(throwable);
+
+ if (!(cause instanceof TakStoppingException) && !(cause instanceof NodeStoppingException)) {
+ LOG.error("Error starting index building: ", cause, indexId);
+ }
+ }
+
+ return null;
+ });
+ } finally {
+ leaveBusy();
+ }
+ }
+
+ /** Stops the execution of a task. */
+ void stop() {
+ if (!taskStopGuard.compareAndSet(false, true)) {
+ return;
+ }
+
+ taskBusyLock.block();
+ }
+
+ private CompletableFuture<Void> awaitActivateForCatalogVersionOfIndexCreation() {
+ return inBusyLock(() -> {
+ Catalog catalog = catalogManager.catalog(catalogVersionOfIndexCreation);
+
+ assert catalog != null : IgniteStringFormatter.format("Missing catalog version: [indexId={}, catalogVersion={}]",
+ indexId, catalogVersionOfIndexCreation);
+
+ return clockWaiter.waitFor(clusterWideEnsuredActivationTimestamp(catalog));
+ });
+ }
+
+ private CompletableFuture<Void> ensureThatLocalNodeStillPrimaryReplica() {
+ return awaitPrimaryReplica().thenAccept(replicaMeta -> {
+ if (!enterBusy()) {
+ throw new TakStoppingException();
+ }
+
+ try {
+ if (!isPrimaryReplica(replicaMeta, localNode(clusterService), clock.now())) {
+ // Lease has expired, we stop the task.
+ throw new TakStoppingException();
+ }
+ } finally {
+ leaveBusy();
+ }
+ });
+ }
+
+ private CompletableFuture<ReplicaMeta> awaitPrimaryReplica() {
+ return inBusyLock(() -> {
+ TablePartitionId groupId = new TablePartitionId(tableId, 0);
+
+ return placementDriver.awaitPrimaryReplica(groupId, clock.now(), AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC, SECONDS)
+ .handle((replicaMeta, throwable) -> {
+ if (throwable != null) {
+ Throwable cause = unwrapCause(throwable);
+
+ if (cause instanceof PrimaryReplicaAwaitTimeoutException) {
+ return awaitPrimaryReplica();
+ }
+
+ return CompletableFuture.<ReplicaMeta>failedFuture(cause);
+ }
+
+ return completedFuture(replicaMeta);
+ }).thenCompose(Function.identity());
+ });
+ }
+
+ private CompletableFuture<Void> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(LogicalTopologySnapshot logicalTopologySnapshot) {
+ return inBusyLock(() -> {
+ Set<LogicalNode> remainingNodes = ConcurrentHashMap.newKeySet();
+ remainingNodes.addAll(logicalTopologySnapshot.nodes());
+
+ var nodeLeftLogicalTopologyListener = new NodeLeftLogicalTopologyListener(remainingNodes);
+ logicalTopologyService.addEventListener(nodeLeftLogicalTopologyListener);
+
+ CompletableFuture[] futures = remainingNodes.stream()
+ .map(node -> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(node, remainingNodes))
+ .toArray(CompletableFuture[]::new);
+
+ return allOf(futures)
+ .whenComplete((unused, throwable) -> logicalTopologyService.removeEventListener(nodeLeftLogicalTopologyListener));
+ });
+ }
+
+ private CompletableFuture<NetworkMessage> awaitFinishRwTxsBeforeCatalogVersionOfIndexCreation(
+ LogicalNode targetNode,
+ Set<LogicalNode> remainingNodes
+ ) {
+ return inBusyLock(() -> {
+ if (!remainingNodes.contains(targetNode)) {
+ // Target node left logical topology, there is no need to do anything.
+ return nullCompletedFuture();
+ }
+
+ IsNodeFinishedRwTransactionsStartedBeforeRequest request = isNodeFinishedRwTransactionsStartedBeforeRequest();
+
+ return clusterService.messagingService().invoke(targetNode, request, INVOKE_MESSAGE_TIMEOUT_MILLS)
+ .handle((message, throwable) -> {
+ if (throwable != null) {
+ Throwable cause = unwrapCause(throwable);
+
+ if (!(cause instanceof TimeoutException) && !(cause instanceof RecipientLeftException)) {
+ return CompletableFuture.<NetworkMessage>failedFuture(cause);
+ }
+ } else if (((IsNodeFinishedRwTransactionsStartedBeforeResponse) message).finished()) {
+ remainingNodes.remove(targetNode);
+
+ return completedFuture(message);
+ }
+
+ // Let's try to send the message again, but after a short time.
+ return inBusyLock(() -> clockWaiter.waitFor(clock.now().addPhysicalTime(RETRY_SEND_MESSAGE_TIMEOUT_MILLS)))
Review Comment:
try fix it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-21115 Wait for completion of transactions started before index appearance [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #3095:
URL: https://github.com/apache/ignite-3/pull/3095#discussion_r1470645194
##########
modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildingStarterTaskTest.java:
##########
@@ -0,0 +1,349 @@
+/*
+ * 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 java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.Executors.newSingleThreadExecutor;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.catalog.CatalogTestUtils.createTestCatalogManager;
+import static org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus.BUILDING;
+import static org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus.REGISTERED;
+import static org.apache.ignite.internal.index.TestIndexManagementUtils.COLUMN_NAME;
+import static org.apache.ignite.internal.index.TestIndexManagementUtils.INDEX_NAME;
+import static org.apache.ignite.internal.index.TestIndexManagementUtils.LOCAL_NODE;
+import static org.apache.ignite.internal.index.TestIndexManagementUtils.LOGICAL_LOCAL_NODE;
+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.createIndex;
+import static org.apache.ignite.internal.index.TestIndexManagementUtils.createTable;
+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.CompletableFutureMatcher.willCompleteSuccessfully;
+import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
+import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mock.Strictness.LENIENT;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+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.CatalogManager;
+import org.apache.ignite.internal.catalog.ClockWaiter;
+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;
+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.HybridClockImpl;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.index.message.IndexMessagesFactory;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.PrimaryReplicaAwaitException;
+import org.apache.ignite.internal.placementdriver.PrimaryReplicaAwaitTimeoutException;
+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.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.MessagingService;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.RecipientLeftException;
+import org.apache.ignite.network.TopologyService;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+/** For {@link IndexBuildingStarterTask} testing. */
+@ExtendWith(MockitoExtension.class)
+public class IndexBuildingStarterTaskTest extends IgniteAbstractTest {
+ private static final IndexMessagesFactory FACTORY = new IndexMessagesFactory();
+
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ private final HybridClock clock = new HybridClockImpl();
+
+ private final ClockWaiter clockWaiter = spy(new ClockWaiter(NODE_NAME, clock));
+
+ private final CatalogManager catalogManager = createTestCatalogManager(NODE_NAME, clockWaiter);
+
+ private final ExecutorService executor = spy(newSingleThreadExecutor());
+
+ private final ClusterService clusterService = createClusterService();
+
+ @Mock
+ private PlacementDriver placementDriver;
+
+ @Mock(strictness = LENIENT)
+ private LogicalTopologyService logicalTopologyService;
+
+ private IndexBuildingStarterTask task;
+
+ private CatalogIndexDescriptor indexDescriptor;
+
+ @BeforeEach
+ void setUp() {
+ assertThat(allOf(clockWaiter.start(), catalogManager.start()), willCompleteSuccessfully());
+
+ createTable(catalogManager, TABLE_NAME, COLUMN_NAME);
+ createIndex(catalogManager, TABLE_NAME, INDEX_NAME, COLUMN_NAME);
+
+ indexDescriptor = indexDescriptor(catalogManager, INDEX_NAME, clock);
+
+ CompletableFuture<ReplicaMeta> localNodeReplicaMetaFuture = completedFuture(
+ createLocalNodeReplicaMeta(HybridTimestamp.MIN_VALUE, HybridTimestamp.MAX_VALUE)
+ );
+
+ when(placementDriver.awaitPrimaryReplica(any(), any(), anyLong(), any())).thenReturn(localNodeReplicaMetaFuture);
+
+ CompletableFuture<LogicalTopologySnapshot> logicalTopologySnapshotFuture = completedFuture(
+ new LogicalTopologySnapshot(1, List.of(LOGICAL_LOCAL_NODE))
+ );
+
+ when(logicalTopologyService.logicalTopologyOnLeader()).thenReturn(logicalTopologySnapshotFuture);
+
+ task = new IndexBuildingStarterTask(
+ indexDescriptor.id(),
+ indexDescriptor.tableId(),
+ catalogManager,
+ placementDriver,
+ clusterService,
+ logicalTopologyService,
+ clock,
+ clockWaiter,
+ executor,
+ busyLock
+ );
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ closeAll(
+ catalogManager::beforeNodeStop,
+ clockWaiter::beforeNodeStop,
+ catalogManager::stop,
+ clockWaiter::stop,
+ task == null ? null : task::stop
+ );
+
+ shutdownAndAwaitTermination(executor, 1, SECONDS);
+ }
+
+ @Test
+ void testSimpleTaskExecution() {
+ clearInvocations(clockWaiter);
Review Comment:
fix it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-21115 Wait for completion of transactions started before index appearance [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #3095:
URL: https://github.com/apache/ignite-3/pull/3095#discussion_r1470637264
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingStarterController.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * 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 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 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.CatalogService;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
+import org.apache.ignite.internal.catalog.events.CatalogEvent;
+import org.apache.ignite.internal.catalog.events.CreateIndexEventParameters;
+import org.apache.ignite.internal.catalog.events.DropIndexEventParameters;
+import org.apache.ignite.internal.close.ManuallyCloseable;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
+import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * Component is responsible for starting and stopping {@link IndexBuildingStarterTask} via {@link IndexBuildingStarter}.
+ *
+ * <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>
+ *
+ * <br><p>Tasks are started and stopped based on events:</p>
+ * <ul>
+ * <li>{@link CatalogEvent#INDEX_CREATE} - task starts for the new index.</li>
+ * <li>{@link CatalogEvent#INDEX_DROP} - 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 it all new indexes, and when the primary replica expires, it stops all tasks for the table.</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>
+ */
+class IndexBuildingStarterController implements ManuallyCloseable {
+ private final CatalogService catalogService;
+
+ private final PlacementDriver placementDriver;
+
+ private final ClusterService clusterService;
+
+ private final IndexBuildingStarter indexBuildingStarter;
+
+ /** Tables IDs for which the local node is the primary replica for the partition with ID {@code 0}. */
+ private final Set<Integer> tableIds = ConcurrentHashMap.newKeySet();
+
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ private final AtomicBoolean closeGuard = new AtomicBoolean();
+
+ IndexBuildingStarterController(
+ CatalogService catalogService,
+ PlacementDriver placementDriver,
+ ClusterService clusterService,
+ IndexBuildingStarter indexBuildingStarter
+ ) {
+ this.catalogService = catalogService;
+ this.placementDriver = placementDriver;
+ this.clusterService = clusterService;
+ this.indexBuildingStarter = indexBuildingStarter;
+
+ addListeners();
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (!closeGuard.compareAndSet(false, true)) {
+ return;
+ }
+
+ busyLock.block();
+
+ indexBuildingStarter.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_DROP, (parameters, exception) -> {
+ if (exception != null) {
+ return failedFuture(exception);
+ }
+
+ return onIndexDrop((DropIndexEventParameters) parameters).thenApply(unused -> false);
+ });
+
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, (parameters, exception) -> {
+ if (exception != null) {
+ return failedFuture(exception);
+ }
+
+ return onPrimaryReplicaElected(parameters).thenApply(unused -> false);
+ });
+ }
+
+ private CompletableFuture<?> onIndexCreate(CreateIndexEventParameters parameters) {
+ return inBusyLockAsync(busyLock, () -> {
+ CatalogIndexDescriptor indexDescriptor = parameters.indexDescriptor();
+
+ if (tableIds.contains(indexDescriptor.tableId())) {
+ // Start building the index only if the local node is the primary replica for the 0 partition of the table for which the
Review Comment:
Started the sentence a little differently
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org