You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ap...@apache.org on 2023/11/22 16:24:21 UTC
(ignite-3) branch main updated: IGNITE-20910 Add test for inserting data after a restart (#2858)
This is an automated email from the ASF dual-hosted git repository.
apolovtsev pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 35b9c25ec8 IGNITE-20910 Add test for inserting data after a restart (#2858)
35b9c25ec8 is described below
commit 35b9c25ec8139f61b6cdd36d0edeabfd20e6b865
Author: Alexander Polovtcev <al...@gmail.com>
AuthorDate: Wed Nov 22 18:24:15 2023 +0200
IGNITE-20910 Add test for inserting data after a restart (#2858)
---
.../internal/placementdriver/LeaseUpdater.java | 26 +++--
.../ignite/internal/replicator/ReplicaManager.java | 2 +-
.../exception/PrimaryReplicaMissException.java | 2 +-
.../internal/runner/app/ItSingleNodeTest.java | 113 +++++++++++++++++++++
.../distributed/storage/InternalTableImpl.java | 38 ++++---
5 files changed, 151 insertions(+), 30 deletions(-)
diff --git a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java
index bdcbb41dfe..60f0da9575 100644
--- a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java
+++ b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java
@@ -364,14 +364,24 @@ public class LeaseUpdater {
or(notExists(key), value(key).eq(leasesCurrent.leasesBytes())),
put(key, renewedValue),
noop()
- ).thenAccept(success -> {
- if (success) {
- for (Map.Entry<ReplicationGroupId, Boolean> e : toBeNegotiated.entrySet()) {
- Lease lease = renewedLeases.get(e.getKey());
- boolean force = e.getValue();
+ ).whenComplete((success, e) -> {
+ if (e != null) {
+ LOG.error("Lease update invocation failed", e);
- leaseNegotiator.negotiate(lease, force);
- }
+ return;
+ }
+
+ if (!success) {
+ LOG.debug("Lease update invocation failed");
+
+ return;
+ }
+
+ for (Map.Entry<ReplicationGroupId, Boolean> entry : toBeNegotiated.entrySet()) {
+ Lease lease = renewedLeases.get(entry.getKey());
+ boolean force = entry.getValue();
+
+ leaseNegotiator.negotiate(lease, force);
}
});
}
@@ -429,7 +439,7 @@ public class LeaseUpdater {
/**
* Checks that the lease is outdated.
- * {@link Lease#EMPTY_LEASE} is always outdated.
+ * {@link Lease#emptyLease} is always outdated.
*
* @param lease Lease.
* @return True when the candidate can be a leaseholder, otherwise false.
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java
index 1879a8859b..e972094e06 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java
@@ -346,7 +346,7 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
}
private static boolean indicatesUnexpectedProblem(Throwable ex) {
- return !(ex instanceof ExpectedReplicationException);
+ return !(unwrapCause(ex) instanceof ExpectedReplicationException);
}
/**
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/exception/PrimaryReplicaMissException.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/exception/PrimaryReplicaMissException.java
index 01967ec92a..adf828c6bc 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/exception/PrimaryReplicaMissException.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/exception/PrimaryReplicaMissException.java
@@ -25,7 +25,7 @@ import org.jetbrains.annotations.Nullable;
/**
* Unchecked exception that is thrown when a replica is not the current primary replica.
*/
-public class PrimaryReplicaMissException extends IgniteInternalException {
+public class PrimaryReplicaMissException extends IgniteInternalException implements ExpectedReplicationException {
private static final long serialVersionUID = 8755220779942651494L;
/**
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItSingleNodeTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItSingleNodeTest.java
new file mode 100644
index 0000000000..1cd3c47e58
--- /dev/null
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItSingleNodeTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.runner.app;
+
+import static org.apache.ignite.internal.network.configuration.NetworkConfigurationSchema.DEFAULT_PORT;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.testNodeName;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgnitionManager;
+import org.apache.ignite.InitParameters;
+import org.apache.ignite.InitParametersBuilder;
+import org.apache.ignite.internal.testframework.IgniteAbstractTest;
+import org.apache.ignite.internal.testframework.TestIgnitionManager;
+import org.apache.ignite.sql.Session;
+import org.apache.ignite.table.KeyValueView;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+
+/**
+ * Integration tests running against a single Ignite node.
+ */
+public class ItSingleNodeTest extends IgniteAbstractTest {
+ private static final String NODE_CONFIG = "network.nodeFinder.netClusterNodes: [ \"localhost:" + DEFAULT_PORT + "\"]";
+
+ private static final String TABLE_NAME = "TEST_TABLE";
+
+ private String nodeName;
+
+ private Ignite ignite;
+
+ @BeforeEach
+ void setUp(TestInfo testInfo) throws Exception {
+ nodeName = testNodeName(testInfo, 0);
+
+ CompletableFuture<Ignite> igniteFuture = startNodeAsync();
+
+ InitParameters parameters = new InitParametersBuilder()
+ .clusterName("cluster")
+ .destinationNodeName(nodeName)
+ .metaStorageNodeNames(List.of(nodeName))
+ .build();
+
+ IgnitionManager.init(parameters);
+
+ ignite = igniteFuture.get(30, TimeUnit.SECONDS);
+ }
+
+ @AfterEach
+ void tearDown() {
+ stopNode();
+ }
+
+ private CompletableFuture<Ignite> startNodeAsync() {
+ return TestIgnitionManager.start(nodeName, NODE_CONFIG, workDir.resolve(nodeName));
+ }
+
+ private Ignite startNode() throws Exception {
+ return startNodeAsync().get(30, TimeUnit.SECONDS);
+ }
+
+ private void stopNode() {
+ IgnitionManager.stop(nodeName);
+ }
+
+ /**
+ * Kind of a stress test that inserts some tuples after a node has been restarted.
+ */
+ @Disabled("https://issues.apache.org/jira/browse/IGNITE-20911")
+ @Test
+ void testManyPutsAfterRestart() throws Exception {
+ String sqlCreate = "CREATE TABLE " + TABLE_NAME + "(\n"
+ + " key int PRIMARY KEY,\n"
+ + " field1 int\n"
+ + ")";
+
+ try (Session session = ignite.sql().createSession()) {
+ session.execute(null, sqlCreate);
+ }
+
+ stopNode();
+ ignite = startNode();
+
+ KeyValueView<Tuple, Tuple> keyValueView = ignite.tables().table(TABLE_NAME).keyValueView();
+
+ Tuple testValue = Tuple.create().set("field1", 239);
+
+ for (int i = 0; i < 2000; i++) {
+ keyValueView.put(null, Tuple.create().set("key", i), testValue);
+ }
+ }
+}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
index 0f71220cc7..396604f00b 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
@@ -72,6 +72,8 @@ import org.apache.ignite.internal.lang.IgniteBiTuple;
import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.lang.IgnitePentaFunction;
import org.apache.ignite.internal.lang.IgniteTriFunction;
+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.ReplicaMeta;
import org.apache.ignite.internal.raft.Peer;
@@ -121,6 +123,8 @@ import org.jetbrains.annotations.TestOnly;
* Storage of table rows.
*/
public class InternalTableImpl implements InternalTable {
+ private static final IgniteLogger LOG = Loggers.forClass(InternalTableImpl.class);
+
/** Cursor id generator. */
private static final AtomicLong CURSOR_ID_GENERATOR = new AtomicLong();
@@ -462,29 +466,23 @@ public class InternalTableImpl implements InternalTable {
boolean full,
@Nullable BiPredicate<R, ReplicaRequest> noWriteChecker
) {
- CompletableFuture<R> result = new CompletableFuture<>();
-
- enlist(partId, tx).thenCompose(
- primaryReplicaAndTerm -> trackingInvoke(tx, partId, mapFunc, full, primaryReplicaAndTerm, noWriteChecker))
- .handle((res0, e) -> {
- if (e != null) {
- if (e.getCause() instanceof PrimaryReplicaMissException && attempts > 0) {
- return enlistWithRetry(tx, partId, mapFunc, attempts - 1, full, noWriteChecker).handle((r2, e2) -> {
- if (e2 != null) {
- return result.completeExceptionally(e2);
- } else {
- return result.complete(r2);
- }
- });
- }
-
- return result.completeExceptionally(e);
+ return enlist(partId, tx)
+ .thenCompose(primaryReplicaAndTerm -> trackingInvoke(tx, partId, mapFunc, full, primaryReplicaAndTerm, noWriteChecker))
+ .handle((response, e) -> {
+ if (e == null) {
+ return completedFuture(response);
}
- return result.complete(res0);
- });
+ if (attempts > 0 && e.getCause() instanceof PrimaryReplicaMissException) {
+ LOG.info("Primary replica for partition {} changed, retrying the request. Remaining attempts: {}",
+ partId, attempts - 1);
- return result;
+ return enlistWithRetry(tx, partId, mapFunc, attempts - 1, full, noWriteChecker);
+ } else {
+ return CompletableFuture.<R>failedFuture(e);
+ }
+ })
+ .thenCompose(Function.identity());
}
/**