You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ma...@apache.org on 2020/04/16 11:01:40 UTC
[cassandra] branch trunk updated (753b40e -> 7855fef)
This is an automated email from the ASF dual-hosted git repository.
marcuse pushed a change to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git.
from 753b40e Fix flaky ConnectionTest#testMessagePurging patch by Yifan Cai; reviewed by Andres de la Pena for CASSANDRA-15338
new 16f639a Duplicate results with DISTINCT queries in mixed mode
new 5774096 Merge branch 'cassandra-2.2' into cassandra-3.0
new 53740bc Merge branch 'cassandra-3.0' into cassandra-3.11
new 7855fef Merge branch 'cassandra-3.11' into trunk
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
CHANGES.txt | 1 +
.../cassandra/distributed/UpgradeableCluster.java | 7 ++
.../distributed/impl/AbstractCluster.java | 11 ++-
.../impl/DelegatingInvokableInstance.java | 3 +-
.../test/RepairCoordinatorNeighbourDown.java | 2 +-
.../upgrade/MixedModeReadRepairTest.java | 32 +++-----
.../cassandra/distributed/upgrade/PagingTest.java | 96 ++++++++++++++++++++++
.../distributed/upgrade/UpgradeTestBase.java | 13 ++-
8 files changed, 140 insertions(+), 25 deletions(-)
create mode 100644 test/distributed/org/apache/cassandra/distributed/upgrade/PagingTest.java
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
[cassandra] 01/01: Merge branch 'cassandra-3.11' into trunk
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
marcuse pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 7855fef42a7ec96dc9828c1737df218e391c8b8f
Merge: 753b40e 53740bc
Author: Marcus Eriksson <ma...@apache.org>
AuthorDate: Thu Apr 16 12:56:15 2020 +0200
Merge branch 'cassandra-3.11' into trunk
CHANGES.txt | 1 +
.../cassandra/distributed/UpgradeableCluster.java | 7 ++
.../distributed/impl/AbstractCluster.java | 11 ++-
.../impl/DelegatingInvokableInstance.java | 3 +-
.../test/RepairCoordinatorNeighbourDown.java | 2 +-
.../upgrade/MixedModeReadRepairTest.java | 32 +++-----
.../cassandra/distributed/upgrade/PagingTest.java | 96 ++++++++++++++++++++++
.../distributed/upgrade/UpgradeTestBase.java | 13 ++-
8 files changed, 140 insertions(+), 25 deletions(-)
diff --cc test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java
index 8a1eff4,71f3f8c..57ace93
--- a/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java
+++ b/test/distributed/org/apache/cassandra/distributed/UpgradeableCluster.java
@@@ -19,9 -19,10 +19,11 @@@
package org.apache.cassandra.distributed;
import java.io.File;
+ import java.io.IOException;
import java.util.List;
+ import java.util.function.Consumer;
+import org.apache.cassandra.distributed.api.ICluster;
import org.apache.cassandra.distributed.api.IInstanceConfig;
import org.apache.cassandra.distributed.impl.AbstractCluster;
import org.apache.cassandra.distributed.api.IUpgradeableInstance;
diff --cc test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
index bcfaaf5,6539747..cdcc39c
--- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
@@@ -155,16 -162,9 +162,16 @@@ public abstract class AbstractCluster<
@Override
public synchronized void startup()
{
+ startup(AbstractCluster.this);
+ }
+
+ public synchronized void startup(ICluster cluster)
+ {
+ if (cluster != AbstractCluster.this)
+ throw new IllegalArgumentException("Only the owning cluster can be used for startup");
if (!isShutdown)
throw new IllegalStateException();
- delegate().startup(cluster);
- delegateForStartup().startup(AbstractCluster.this);
++ delegateForStartup().startup(cluster);
isShutdown = false;
updateMessagingVersions();
}
diff --cc test/distributed/org/apache/cassandra/distributed/test/RepairCoordinatorNeighbourDown.java
index db01b13,0000000..dd6e2c4
mode 100644,000000..100644
--- a/test/distributed/org/apache/cassandra/distributed/test/RepairCoordinatorNeighbourDown.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/RepairCoordinatorNeighbourDown.java
@@@ -1,199 -1,0 +1,199 @@@
+/*
+ * 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.cassandra.distributed.test;
+
+import java.net.UnknownHostException;
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.google.common.util.concurrent.Uninterruptibles;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.api.NodeToolResult;
+import org.apache.cassandra.distributed.test.DistributedRepairUtils.RepairParallelism;
+import org.apache.cassandra.distributed.test.DistributedRepairUtils.RepairType;
+import org.apache.cassandra.gms.FailureDetector;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.net.Verb;
+import org.apache.cassandra.utils.FBUtilities;
+
+import static java.lang.String.format;
+import static org.apache.cassandra.distributed.api.IMessageFilters.Matcher.of;
+import static org.apache.cassandra.distributed.test.DistributedRepairUtils.assertParentRepairFailedWithMessageContains;
+import static org.apache.cassandra.distributed.test.DistributedRepairUtils.assertParentRepairNotExist;
+import static org.apache.cassandra.distributed.test.DistributedRepairUtils.getRepairExceptions;
+import static org.apache.cassandra.utils.AssertUtil.assertTimeoutPreemptively;
+
+public abstract class RepairCoordinatorNeighbourDown extends RepairCoordinatorBase
+{
+ public RepairCoordinatorNeighbourDown(RepairType repairType, RepairParallelism parallelism, boolean withNotifications)
+ {
+ super(repairType, parallelism, withNotifications);
+ }
+
+ @Before
+ public void beforeTest()
+ {
+ CLUSTER.filters().reset();
+ CLUSTER.forEach(i -> {
+ try
+ {
+ i.startup();
+ }
+ catch (IllegalStateException e)
+ {
+ // ignore, node wasn't down
+ }
+ });
+ }
+
+ @Test
+ public void neighbourDown()
+ {
+ String table = tableName("neighbourdown");
+ assertTimeoutPreemptively(Duration.ofMinutes(1), () -> {
+ CLUSTER.schemaChange(format("CREATE TABLE %s.%s (key text, value text, PRIMARY KEY (key))", KEYSPACE, table));
- Future<Void> shutdownFuture = CLUSTER.get(2).shutdown();
+ String downNodeAddress = CLUSTER.get(2).callOnInstance(() -> FBUtilities.getBroadcastAddressAndPort().toString());
++ Future<Void> shutdownFuture = CLUSTER.get(2).shutdown();
+ try
+ {
+ // wait for the node to stop
+ shutdownFuture.get();
+ // wait for the failure detector to detect this
+ CLUSTER.get(1).runOnInstance(() -> {
+ InetAddressAndPort neighbor;
+ try
+ {
+ neighbor = InetAddressAndPort.getByName(downNodeAddress);
+ }
+ catch (UnknownHostException e)
+ {
+ throw new RuntimeException(e);
+ }
+ while (FailureDetector.instance.isAlive(neighbor))
+ Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
+ });
+
+ long repairExceptions = getRepairExceptions(CLUSTER, 1);
+ NodeToolResult result = repair(1, KEYSPACE, table);
+ result.asserts()
+ .failure()
+ .errorContains("Endpoint not alive");
+ if (withNotifications)
+ {
+ result.asserts()
+ .notificationContains(NodeToolResult.ProgressEventType.START, "Starting repair command")
+ .notificationContains(NodeToolResult.ProgressEventType.START, "repairing keyspace " + KEYSPACE + " with repair options")
+ .notificationContains(NodeToolResult.ProgressEventType.ERROR, "Endpoint not alive")
+ .notificationContains(NodeToolResult.ProgressEventType.COMPLETE, "finished with error");
+ }
+
+ Assert.assertEquals(repairExceptions + 1, getRepairExceptions(CLUSTER, 1));
+ }
+ finally
+ {
+ CLUSTER.get(2).startup();
+ }
+
+ // make sure to call outside of the try/finally so the node is up so we can actually query
+ if (repairType != RepairType.PREVIEW)
+ {
+ assertParentRepairFailedWithMessageContains(CLUSTER, KEYSPACE, table, "Endpoint not alive");
+ }
+ else
+ {
+ assertParentRepairNotExist(CLUSTER, KEYSPACE, table);
+ }
+ });
+ }
+
+ @Test
+ public void validationParticipentCrashesAndComesBack()
+ {
+ // Test what happens when a participant restarts in the middle of validation
+ // Currently this isn't recoverable but could be.
+ // TODO since this is a real restart, how would I test "long pause"? Can't send SIGSTOP since same procress
+ String table = tableName("validationparticipentcrashesandcomesback");
+ assertTimeoutPreemptively(Duration.ofMinutes(1), () -> {
+ CLUSTER.schemaChange(format("CREATE TABLE %s.%s (key text, value text, PRIMARY KEY (key))", KEYSPACE, table));
+ AtomicReference<Future<Void>> participantShutdown = new AtomicReference<>();
+ CLUSTER.verbs(Verb.VALIDATION_REQ).to(2).messagesMatching(of(m -> {
+ // the nice thing about this is that this lambda is "capturing" and not "transfer", what this means is that
+ // this lambda isn't serialized and any object held isn't copied.
+ participantShutdown.set(CLUSTER.get(2).shutdown());
+ return true; // drop it so this node doesn't reply before shutdown.
+ })).drop();
+ // since nodetool is blocking, need to handle participantShutdown in the background
+ CompletableFuture<Void> recovered = CompletableFuture.runAsync(() -> {
+ try {
+ while (participantShutdown.get() == null) {
+ // event not happened, wait for it
+ TimeUnit.MILLISECONDS.sleep(100);
+ }
+ Future<Void> f = participantShutdown.get();
+ f.get(); // wait for shutdown to complete
+ CLUSTER.get(2).startup();
+ } catch (Exception e) {
+ if (e instanceof RuntimeException) {
+ throw (RuntimeException) e;
+ }
+ throw new RuntimeException(e);
+ }
+ });
+
+ long repairExceptions = getRepairExceptions(CLUSTER, 1);
+ NodeToolResult result = repair(1, KEYSPACE, table);
+ recovered.join(); // if recovery didn't happen then the results are not what are being tested, so block here first
+ result.asserts()
+ .failure();
+ if (withNotifications)
+ {
+ result.asserts()
+ .errorContains("Endpoint 127.0.0.2:7012 died")
+ .notificationContains(NodeToolResult.ProgressEventType.ERROR, "Endpoint 127.0.0.2:7012 died")
+ .notificationContains(NodeToolResult.ProgressEventType.COMPLETE, "finished with error");
+ }
+ else
+ {
+ // Right now coordination doesn't propgate the first exception, so we only know "there exists a issue".
+ // With notifications on nodetool will see the error then complete, so the cmd state (what nodetool
+ // polls on) is ignored. With notifications off, the poll await fails and queries cmd state, and that
+ // will have the below error.
+ // NOTE: this isn't desireable, would be good to propgate
+ result.asserts()
+ .errorContains("Some repair failed");
+ }
+
+ Assert.assertEquals(repairExceptions + 1, getRepairExceptions(CLUSTER, 1));
+ if (repairType != RepairType.PREVIEW)
+ {
+ assertParentRepairFailedWithMessageContains(CLUSTER, KEYSPACE, table, "Endpoint 127.0.0.2:7012 died");
+ }
+ else
+ {
+ assertParentRepairNotExist(CLUSTER, KEYSPACE, table);
+ }
+ });
+ }
+}
diff --cc test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeReadRepairTest.java
index 6df72b8,cc50053..babfdbc
--- a/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeReadRepairTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/upgrade/MixedModeReadRepairTest.java
@@@ -33,26 -32,21 +32,21 @@@ public class MixedModeReadRepairTest ex
new TestCase()
.nodes(2)
.upgrade(Versions.Major.v22, Versions.Major.v30)
- .nodesToUpgrade(2)
- .setup((cluster) -> cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk ascii, b boolean, v blob, PRIMARY KEY (pk)) WITH COMPACT STORAGE"))
- .runAfterClusterUpgrade((cluster) -> {
- // now node2 is 3.0 and node1 is 2.2
+ .setup((cluster) -> cluster.schemaChange("CREATE TABLE " + DistributedTestBase.KEYSPACE + ".tbl (pk ascii, b boolean, v blob, PRIMARY KEY (pk)) WITH COMPACT STORAGE"))
+ .runAfterNodeUpgrade((cluster, node) -> {
+ if (node != 1)
+ return;
+ // now node1 is 3.0 and node2 is 2.2
// make sure 2.2 side does not get the mutation
- cluster.get(2).executeInternal("DELETE FROM " + KEYSPACE + ".tbl WHERE pk = ?",
- "something");
+ cluster.get(1).executeInternal("DELETE FROM " + DistributedTestBase.KEYSPACE + ".tbl WHERE pk = ?",
+ "something");
// trigger a read repair
- cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = ?",
+ cluster.coordinator(2).execute("SELECT * FROM " + DistributedTestBase.KEYSPACE + ".tbl WHERE pk = ?",
ConsistencyLevel.ALL,
"something");
- cluster.get(1).flush(KEYSPACE);
- // upgrade node1 to 3.0
- cluster.get(1).shutdown().get();
- Versions allVersions = find();
- cluster.get(1).setVersion(allVersions.getLatest(Versions.Major.v30));
- cluster.get(1).startup();
-
- // and make sure the sstables are readable
- cluster.get(1).forceCompact(KEYSPACE, "tbl");
- }).run();
+ cluster.get(2).flush(DistributedTestBase.KEYSPACE);
+ })
+ .runAfterClusterUpgrade((cluster) -> cluster.get(2).forceCompact(DistributedTestBase.KEYSPACE, "tbl"))
+ .run();
}
-}
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org