natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix StreamingMetricTests flakiness 

patch by Benjamin Lerer; reviewed by Ekaterina Dimitrova for CASSANDRA-16598
This commit is contained in:
Benjamin Lerer 2021-06-10 14:55:07 +02:00
parent 23ad7c301e
commit 14fd0ad4e3
1 changed files with 17 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
test/distributed/org/apache/cassandra/distributed/test/metrics/StreamingMetricsTest.java
View File

@ -19,7 +19,6 @@
package org.apache.cassandra.distributed.test.metrics;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import org.apache.cassandra.db.ColumnFamilyStore;
@ -28,6 +27,7 @@ import org.junit.Test;
import org.apache.cassandra.db.SystemKeyspace;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.IMessageFilters;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.metrics.StreamingMetrics;
@ -35,6 +35,7 @@ import org.apache.cassandra.metrics.StreamingMetrics;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.assertj.core.api.Assertions.assertThat;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.apache.cassandra.net.Verb.MUTATION_REQ;
public class StreamingMetricsTest extends TestBaseImpl
{
@ -69,7 +70,7 @@ public class StreamingMetricsTest extends TestBaseImpl
{
cluster.schemaChange(String.format("CREATE TABLE %s.cf (k text, c1 text, c2 text, PRIMARY KEY (k)) WITH compaction = {'class': '%s', 'enabled': 'false'}", KEYSPACE, "LeveledCompactionStrategy"));
cluster.get(3).shutdown().get(10, TimeUnit.SECONDS);
IMessageFilters.Filter drop1to3 = cluster.filters().verbs(MUTATION_REQ.id).from(1).to(3).drop();
final int rowsPerFile = 500;
final int files = 5;
@ -81,14 +82,14 @@ public class StreamingMetricsTest extends TestBaseImpl
ConsistencyLevel.ONE,
Integer.toString(i));
}
cluster.get(1).nodetool("flush");
cluster.get(2).nodetool("flush");
cluster.get(1).flush(KEYSPACE);
cluster.get(2).flush(KEYSPACE);
}
cluster.get(3).startup();
drop1to3.off();
// Checks that the table is empty on node 3
Object[][] results = cluster.get(3).executeInternal(String.format("SELECT k, c1, c2 FROM %s.cf;", KEYSPACE));
Object[][] results = cluster.get(3).executeInternal(withKeyspace("SELECT k, c1, c2 FROM %s.cf;"));
assertThat(results.length).isEqualTo(0);
checkThatNoStreamingOccuredBetweenTheThreeNodes(cluster);
@ -163,9 +164,11 @@ public class StreamingMetricsTest extends TestBaseImpl
final int rowsPerFile = 500;
final int files = 5;
cluster.get(3).shutdown().get(10, TimeUnit.SECONDS);
cluster.get(1).nodetool("disableautocompaction", KEYSPACE);
cluster.get(2).nodetool("disableautocompaction", KEYSPACE);
cluster.get(3).nodetool("disableautocompaction", KEYSPACE);
IMessageFilters.Filter drop1to3 = cluster.filters().verbs(MUTATION_REQ.id).from(1).to(3).drop();
int sstablesInitiallyOnNode2 = 0;
int sstablesInitiallyOnNode3 = 0;
@ -178,15 +181,14 @@ public class StreamingMetricsTest extends TestBaseImpl
ConsistencyLevel.ONE,
Integer.toString(i));
}
cluster.get(1).nodetool("flush");
cluster.get(2).nodetool("flush");
cluster.get(1).flush(KEYSPACE);
cluster.get(2).flush(KEYSPACE);
sstablesInitiallyOnNode2++;
}
cluster.get(3).startup();
cluster.get(3).nodetool("disableautocompaction", KEYSPACE);
drop1to3.off();
cluster.get(2).shutdown().get(10, TimeUnit.SECONDS);
IMessageFilters.Filter drop1to2 = cluster.filters().verbs(MUTATION_REQ.id).from(1).to(2).drop();
for (int k = 3; k < files; k++)
{
@ -196,13 +198,12 @@ public class StreamingMetricsTest extends TestBaseImpl
ConsistencyLevel.ONE,
Integer.toString(i));
}
cluster.get(1).nodetool("flush");
cluster.get(3).nodetool("flush");
cluster.get(1).flush(KEYSPACE);
cluster.get(3).flush(KEYSPACE);
sstablesInitiallyOnNode3++;
}
cluster.get(2).startup();
cluster.get(2).nodetool("disableautocompaction", KEYSPACE);
drop1to2.off();
checkThatNoStreamingOccuredBetweenTheThreeNodes(cluster);