diff --git a/NEWS.txt b/NEWS.txt index 5c814cddb7..2b45a2a459 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -52,6 +52,13 @@ Upgrading for further details. You also need to regenerate passwords for users for who the password was created while the above property was set to be more than 30 otherwise they will not be able to log in. +Statement re-prepare storms +--------------------------- + - CASSANDRA-15252 has changed how prepared statement ids are computed in order to avoid infinite re-prepare + loops caused by the driver. This new behaviour will be picked up only when the entire cluster has been updated + to 3.0.26 or higher. In case of a mixed version cluster, different major versions will be taken as a minimum + required version. + 3.11.11 ======= diff --git a/src/java/org/apache/cassandra/cql3/QueryProcessor.java b/src/java/org/apache/cassandra/cql3/QueryProcessor.java index 0c607352da..589a25a8c9 100644 --- a/src/java/org/apache/cassandra/cql3/QueryProcessor.java +++ b/src/java/org/apache/cassandra/cql3/QueryProcessor.java @@ -47,6 +47,7 @@ import org.apache.cassandra.db.partitions.PartitionIterator; import org.apache.cassandra.db.partitions.PartitionIterators; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.exceptions.*; +import org.apache.cassandra.gms.Gossiper; import org.apache.cassandra.metrics.CQLMetrics; import org.apache.cassandra.service.*; import org.apache.cassandra.service.pager.QueryPager; @@ -62,6 +63,17 @@ public class QueryProcessor implements QueryHandler { public static final CassandraVersion CQL_VERSION = new CassandraVersion("3.4.4"); + /** + * If a query is prepared with a fully qualified name, but the user also uses USE (specifically when USE keyspace + * is different) then the IDs generated could change over time; invalidating the assumption that IDs won't ever + * change. In the version defined below, the USE keyspace is ignored when a fully-qualified name is used as an + * attempt to make IDs stable. + */ + private static final CassandraVersion PREPARE_ID_BEHAVIOR_CHANGE_30 = new CassandraVersion("3.0.26"); + private static final CassandraVersion PREPARE_ID_BEHAVIOR_CHANGE_3X = new CassandraVersion("3.11.12"); + private static final CassandraVersion PREPARE_ID_BEHAVIOR_CHANGE_40 = new CassandraVersion("4.0.1"); + + public static final QueryProcessor instance = new QueryProcessor(); private static final Logger logger = LoggerFactory.getLogger(QueryProcessor.class); @@ -425,7 +437,29 @@ public class QueryProcessor implements QueryHandler throw new InvalidRequestException(String.format("Too many markers(?). %d markers exceed the allowed maximum of %d", boundTerms, FBUtilities.MAX_UNSIGNED_SHORT)); assert boundTerms == prepared.boundNames.size(); - return storePreparedStatement(queryString, clientState.getRawKeyspace(), prepared, forThrift); + if (prepared.keyspace != null) + { + // Edge-case of CASSANDRA-15252 in mixed-mode cluster. We accept that 15252 itself can manifest in a + // cluster that has both old and new nodes, but we would like to avoid a situation when the fix adds + // a new behaviour that can break which, in addition, can get triggered more frequently. + // If statement ID was generated on the old node _with_ use, when attempting to execute on the new node, + // we may fall into infinite loop. To break out of this loop, we put a prepared statement that client + // expects into cache, so that it could get PREPARED response on the second try. + ResultMessage.Prepared newBehavior = storePreparedStatement(queryString, null, prepared, forThrift); + ResultMessage.Prepared oldBehavior = clientState.getRawKeyspace() != null ? storePreparedStatement(queryString, clientState.getRawKeyspace(), prepared, forThrift) : newBehavior; + CassandraVersion minVersion = Gossiper.instance.getMinVersion(20, TimeUnit.MILLISECONDS); + + // Default to old behaviour in case we're not sure about the version. Even if we ever flip back to the old + // behaviour due to the gossip bug or incorrect version string, we'll end up with two re-prepare round-trips. + return minVersion != null && + ((minVersion.major == 3 && minVersion.minor == 0 && minVersion.compareTo(PREPARE_ID_BEHAVIOR_CHANGE_30) >= 0) || + (minVersion.major == 3 && minVersion.minor != 0 && minVersion.compareTo(PREPARE_ID_BEHAVIOR_CHANGE_3X) >= 0) || + (minVersion.major == 4 && minVersion.compareTo(PREPARE_ID_BEHAVIOR_CHANGE_40) >= 0)) ? newBehavior : oldBehavior; + } + else + { + return storePreparedStatement(queryString, clientState.getRawKeyspace(), prepared, forThrift); + } } private static MD5Digest computeId(String queryString, String keyspace) @@ -440,12 +474,13 @@ public class QueryProcessor implements QueryHandler return toHash.hashCode(); } - private static ResultMessage.Prepared getStoredPreparedStatement(String queryString, String keyspace, boolean forThrift) + @VisibleForTesting + public static ResultMessage.Prepared getStoredPreparedStatement(String queryString, String clientKeyspace, boolean forThrift) throws InvalidRequestException { if (forThrift) { - Integer thriftStatementId = computeThriftId(queryString, keyspace); + Integer thriftStatementId = computeThriftId(queryString, clientKeyspace); ParsedStatement.Prepared existing = thriftPreparedStatements.get(thriftStatementId); if (existing == null) return null; @@ -456,7 +491,7 @@ public class QueryProcessor implements QueryHandler } else { - MD5Digest statementId = computeId(queryString, keyspace); + MD5Digest statementId = computeId(queryString, clientKeyspace); ParsedStatement.Prepared existing = preparedStatements.get(statementId); if (existing == null) return null; @@ -467,7 +502,8 @@ public class QueryProcessor implements QueryHandler } } - private static ResultMessage.Prepared storePreparedStatement(String queryString, String keyspace, ParsedStatement.Prepared prepared, boolean forThrift) + @VisibleForTesting + public static ResultMessage.Prepared storePreparedStatement(String queryString, String keyspace, ParsedStatement.Prepared prepared, boolean forThrift) throws InvalidRequestException { // Concatenate the current keyspace so we don't mix prepared statements between keyspace (#5352). @@ -620,6 +656,13 @@ public class QueryProcessor implements QueryHandler internalStatements.clear(); } + @VisibleForTesting + public static void clearPreparedStatementsCache() + { + preparedStatements.clear(); + thriftPreparedStatements.clear(); + } + private static class MigrationSubscriber extends MigrationListener { private static void removeInvalidPreparedStatements(String ksName, String cfName) diff --git a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java index caf8c97cc9..e4cca2a220 100644 --- a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java @@ -561,7 +561,7 @@ public class BatchStatement implements CQLStatement short[] partitionKeyBindIndexes = (haveMultipleCFs || batchStatement.statements.isEmpty())? null : boundNames.getPartitionKeyBindIndexes(batchStatement.statements.get(0).cfm); - return new ParsedStatement.Prepared(batchStatement, boundNames, partitionKeyBindIndexes); + return new ParsedStatement.Prepared(batchStatement, boundNames, partitionKeyBindIndexes, null); } } diff --git a/src/java/org/apache/cassandra/cql3/statements/CFStatement.java b/src/java/org/apache/cassandra/cql3/statements/CFStatement.java index 9b2987cb9d..f3cdd7f64d 100644 --- a/src/java/org/apache/cassandra/cql3/statements/CFStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/CFStatement.java @@ -37,7 +37,7 @@ public abstract class CFStatement extends ParsedStatement { if (!cfName.hasKeyspace()) { - // XXX: We explicitely only want to call state.getKeyspace() in this case, as we don't want to throw + // XXX: We explicitly only want to call state.getKeyspace() in this case, as we don't want to throw // if not logged in any keyspace but a keyspace is explicitely set on the statement. So don't move // the call outside the 'if' or replace the method by 'prepareKeyspace(state.getKeyspace())' cfName.setKeyspace(state.getKeyspace(), true); diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java index d2e693ac24..8376a0a89d 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java @@ -814,7 +814,7 @@ public abstract class ModificationStatement implements CQLStatement { VariableSpecifications boundNames = getBoundVariables(); ModificationStatement statement = prepare(boundNames, clientState); - return new ParsedStatement.Prepared(statement, boundNames, boundNames.getPartitionKeyBindIndexes(statement.cfm)); + return new ParsedStatement.Prepared(statement, boundNames, boundNames.getPartitionKeyBindIndexes(statement.cfm), statement.cfm.ksName); } public ModificationStatement prepare(VariableSpecifications boundNames, ClientState clientState) diff --git a/src/java/org/apache/cassandra/cql3/statements/ParsedStatement.java b/src/java/org/apache/cassandra/cql3/statements/ParsedStatement.java index aa292f5266..dcc476fbca 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ParsedStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ParsedStatement.java @@ -20,6 +20,8 @@ package org.apache.cassandra.cql3.statements; import java.util.Collections; import java.util.List; +import javax.annotation.Nullable; + import org.apache.cassandra.cql3.*; import org.apache.cassandra.cql3.functions.Function; import org.apache.cassandra.exceptions.RequestValidationException; @@ -59,24 +61,34 @@ public abstract class ParsedStatement public final CQLStatement statement; public final List boundNames; + + @Nullable public final short[] partitionKeyBindIndexes; - protected Prepared(CQLStatement statement, List boundNames, short[] partitionKeyBindIndexes) + @Nullable + public final String keyspace; + + protected Prepared(CQLStatement statement, List boundNames, short[] partitionKeyBindIndexes, String keyspace) { this.statement = statement; this.boundNames = boundNames; this.partitionKeyBindIndexes = partitionKeyBindIndexes; - this.rawCQLStatement = ""; + this.keyspace = keyspace; } public Prepared(CQLStatement statement, VariableSpecifications names, short[] partitionKeyBindIndexes) { - this(statement, names.getSpecifications(), partitionKeyBindIndexes); + this(statement, names, partitionKeyBindIndexes, null); + } + + public Prepared(CQLStatement statement, VariableSpecifications names, short[] partitionKeyBindIndexes, String keyspace) + { + this(statement, names.getSpecifications(), partitionKeyBindIndexes, keyspace); } public Prepared(CQLStatement statement) { - this(statement, Collections.emptyList(), null); + this(statement, Collections.emptyList(), null, null); } } diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java index ffae8468e7..632bf94f9c 100644 --- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java @@ -1008,7 +1008,12 @@ public class SelectStatement implements CQLStatement prepareLimit(boundNames, limit, keyspace(), limitReceiver()), prepareLimit(boundNames, perPartitionLimit, keyspace(), perPartitionLimitReceiver())); - return new ParsedStatement.Prepared(stmt, boundNames, boundNames.getPartitionKeyBindIndexes(cfm)); + return prepare(stmt, boundNames, cfm); + } + + protected ParsedStatement.Prepared prepare(SelectStatement stmt, VariableSpecifications boundNames, CFMetaData cfm) + { + return new ParsedStatement.Prepared(stmt, boundNames, boundNames.getPartitionKeyBindIndexes(cfm), cfm.ksName); } /** diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index 08cd106277..363db5fd3f 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -25,11 +25,15 @@ import java.util.concurrent.*; import java.util.concurrent.locks.ReentrantLock; import javax.annotation.Nullable; import java.util.stream.Collectors; +import java.util.stream.Stream; + +import javax.annotation.Nullable; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Throwables; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Iterables; import com.google.common.collect.Sets; import com.google.common.util.concurrent.ListenableFutureTask; import com.google.common.util.concurrent.Uninterruptibles; @@ -58,6 +62,8 @@ import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.CassandraVersion; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.JVMStabilityInspector; +import org.apache.cassandra.utils.RecomputingSupplier; + import static org.apache.cassandra.utils.ExecutorUtils.awaitTermination; import static org.apache.cassandra.utils.ExecutorUtils.shutdown; @@ -253,6 +259,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean } } + private final RecomputingSupplier minVersionSupplier = new RecomputingSupplier<>(this::computeMinVersion, executor); + private Gossiper() { // half of QUARATINE_DELAY, to ensure justRemovedEndpoints has enough leeway to prevent re-gossip @@ -262,6 +270,32 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean // Register this instance with JMX MBeanWrapper.instance.registerMBean(this, MBEAN_NAME); + + + subscribers.add(new IEndpointStateChangeSubscriber() + { + public void onJoin(InetAddress endpoint, EndpointState state) + { + maybeRecompute(state); + } + + public void onAlive(InetAddress endpoint, EndpointState state) + { + maybeRecompute(state); + } + + private void maybeRecompute(EndpointState state) + { + if (state.getApplicationState(ApplicationState.RELEASE_VERSION) != null) + minVersionSupplier.recompute(); + } + + public void onChange(InetAddress endpoint, ApplicationState state, VersionedValue value) + { + if (state == ApplicationState.RELEASE_VERSION) + minVersionSupplier.recompute(); + } + }); } public void setLastProcessedMessageAt(long timeInMillis) @@ -1474,6 +1508,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean maybeInitializeLocalState(generationNbr); EndpointState localState = endpointStateMap.get(FBUtilities.getBroadcastAddress()); localState.addApplicationStates(preloadLocalStates); + minVersionSupplier.recompute(); //notify snitches that Gossiper is about to start DatabaseDescriptor.getEndpointSnitch().gossiperStarting(); @@ -1872,6 +1907,72 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean ExecutorUtils.shutdownAndWait(timeout, unit, executor); } + @Nullable + public CassandraVersion getMinVersion(int delay, TimeUnit timeUnit) + { + try + { + return minVersionSupplier.get(delay, timeUnit); + } + catch (TimeoutException e) + { + // Timeouts here are harmless: they won't cause reprepares and may only + // cause the old version of the hash to be kept for longer + return null; + } + catch (Throwable e) + { + logger.error("Caught an exception while waiting for min version", e); + return null; + } + } + + @Nullable + private String getReleaseVersionString(InetAddress ep) + { + EndpointState state = getEndpointStateForEndpoint(ep); + if (state == null) + return null; + + VersionedValue value = state.getApplicationState(ApplicationState.RELEASE_VERSION); + return value == null ? null : value.value; + } + + private CassandraVersion computeMinVersion() + { + CassandraVersion minVersion = null; + + for (InetAddress addr : Iterables.concat(Gossiper.instance.getLiveMembers(), + Gossiper.instance.getUnreachableMembers())) + { + String versionString = getReleaseVersionString(addr); + // Raced with changes to gossip state, wait until next iteration + if (versionString == null) + return null; + + CassandraVersion version; + + try + { + version = new CassandraVersion(versionString); + } + catch (Throwable t) + { + JVMStabilityInspector.inspectThrowable(t); + String message = String.format("Can't parse version string %s", versionString); + logger.warn(message); + if (logger.isDebugEnabled()) + logger.debug(message, t); + return null; + } + + if (minVersion == null || version.compareTo(minVersion) < 0) + minVersion = version; + } + + return minVersion; + } + @VisibleForTesting public void setAnyNodeOn30(boolean anyNodeOn30) { diff --git a/src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java b/src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java index 1bfd678cbe..861d4ac63e 100644 --- a/src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java +++ b/src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java @@ -36,17 +36,17 @@ public interface IEndpointStateChangeSubscriber * @param endpoint endpoint for which the state change occurred. * @param epState state that actually changed for the above endpoint. */ - public void onJoin(InetAddress endpoint, EndpointState epState); + default void onJoin(InetAddress endpoint, EndpointState epState) {} - public void beforeChange(InetAddress endpoint, EndpointState currentState, ApplicationState newStateKey, VersionedValue newValue); + default void beforeChange(InetAddress endpoint, EndpointState currentState, ApplicationState newStateKey, VersionedValue newValue) {} - public void onChange(InetAddress endpoint, ApplicationState state, VersionedValue value); + default void onChange(InetAddress endpoint, ApplicationState state, VersionedValue value) {} - public void onAlive(InetAddress endpoint, EndpointState state); + default void onAlive(InetAddress endpoint, EndpointState state) {} - public void onDead(InetAddress endpoint, EndpointState state); + default void onDead(InetAddress endpoint, EndpointState state) {} - public void onRemove(InetAddress endpoint); + default void onRemove(InetAddress endpoint) {} /** * Called whenever a node is restarted. @@ -54,5 +54,5 @@ public interface IEndpointStateChangeSubscriber * previously marked down. It will have only if {@code state.isAlive() == false} * as {@code state} is from before the restarted node is marked up. */ - public void onRestart(InetAddress endpoint, EndpointState state); + default void onRestart(InetAddress endpoint, EndpointState state) {} } diff --git a/src/java/org/apache/cassandra/repair/RepairSession.java b/src/java/org/apache/cassandra/repair/RepairSession.java index c9376cbc17..1207d36316 100644 --- a/src/java/org/apache/cassandra/repair/RepairSession.java +++ b/src/java/org/apache/cassandra/repair/RepairSession.java @@ -333,12 +333,6 @@ public class RepairSession extends AbstractFuture implement terminate(); } - public void onJoin(InetAddress endpoint, EndpointState epState) {} - public void beforeChange(InetAddress endpoint, EndpointState currentState, ApplicationState newStateKey, VersionedValue newValue) {} - public void onChange(InetAddress endpoint, ApplicationState state, VersionedValue value) {} - public void onAlive(InetAddress endpoint, EndpointState state) {} - public void onDead(InetAddress endpoint, EndpointState state) {} - public void onRemove(InetAddress endpoint) { convict(endpoint, Double.MAX_VALUE); diff --git a/src/java/org/apache/cassandra/service/LoadBroadcaster.java b/src/java/org/apache/cassandra/service/LoadBroadcaster.java index 28e1ae8490..0a3e1a4fbd 100644 --- a/src/java/org/apache/cassandra/service/LoadBroadcaster.java +++ b/src/java/org/apache/cassandra/service/LoadBroadcaster.java @@ -60,14 +60,6 @@ public class LoadBroadcaster implements IEndpointStateChangeSubscriber onChange(endpoint, ApplicationState.LOAD, localValue); } } - - public void beforeChange(InetAddress endpoint, EndpointState currentState, ApplicationState newStateKey, VersionedValue newValue) {} - - public void onAlive(InetAddress endpoint, EndpointState state) {} - - public void onDead(InetAddress endpoint, EndpointState state) {} - - public void onRestart(InetAddress endpoint, EndpointState state) {} public void onRemove(InetAddress endpoint) { diff --git a/src/java/org/apache/cassandra/streaming/StreamSession.java b/src/java/org/apache/cassandra/streaming/StreamSession.java index 08cd22d80e..675304fc10 100644 --- a/src/java/org/apache/cassandra/streaming/StreamSession.java +++ b/src/java/org/apache/cassandra/streaming/StreamSession.java @@ -749,12 +749,6 @@ public class StreamSession implements IEndpointStateChangeSubscriber maybeCompleted(); } - public void onJoin(InetAddress endpoint, EndpointState epState) {} - public void beforeChange(InetAddress endpoint, EndpointState currentState, ApplicationState newStateKey, VersionedValue newValue) {} - public void onChange(InetAddress endpoint, ApplicationState state, VersionedValue value) {} - public void onAlive(InetAddress endpoint, EndpointState state) {} - public void onDead(InetAddress endpoint, EndpointState state) {} - public void onRemove(InetAddress endpoint) { logger.error("[Stream #{}] Session failed because remote peer {} has left.", planId(), peer.getHostAddress()); diff --git a/src/java/org/apache/cassandra/utils/RecomputingSupplier.java b/src/java/org/apache/cassandra/utils/RecomputingSupplier.java new file mode 100644 index 0000000000..ba6a1ff55d --- /dev/null +++ b/src/java/org/apache/cassandra/utils/RecomputingSupplier.java @@ -0,0 +1,122 @@ +/* + * 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.utils; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; + +/** + * Supplier that caches the last computed value until it is reset, forcing every caller of + * {@link RecomputingSupplier#get(long, TimeUnit)} to wait until this value is computed if + * it was not computed yet. + * + * Calling {@link RecomputingSupplier#recompute()} won't reset value for the already + * waiting consumers, but instead will schedule one recomputation as soon as current one is done. + */ +public class RecomputingSupplier +{ + private final Supplier supplier; + private final AtomicReference> cached = new AtomicReference<>(null); + private final AtomicBoolean workInProgress = new AtomicBoolean(false); + private final ExecutorService executor; + + public RecomputingSupplier(Supplier supplier, ExecutorService executor) + { + this.supplier = supplier; + this.executor = executor; + } + + public void recompute() + { + CompletableFuture current = cached.get(); + boolean origWip = workInProgress.get(); + + if (origWip || (current != null && !current.isDone())) + { + if (cached.get() != current) + executor.submit(this::recompute); + return; // if work is has not started yet, schedule task for the future + } + + assert current == null || current.isDone(); + + // The work is not in progress, and current future is done. Try to submit a new task. + CompletableFuture lazyValue = new CompletableFuture<>(); + if (cached.compareAndSet(current, lazyValue)) + executor.submit(() -> doWork(lazyValue)); + else + executor.submit(this::recompute); // Lost CAS, resubmit + } + + private void doWork(CompletableFuture lazyValue) + { + T value = null; + Throwable err = null; + try + { + sanityCheck(workInProgress.compareAndSet(false, true)); + value = supplier.get(); + } + catch (Throwable t) + { + err = t; + } + finally + { + sanityCheck(workInProgress.compareAndSet(true, false)); + } + + if (err == null) + lazyValue.complete(value); + else + lazyValue.completeExceptionally(err); + } + + private static void sanityCheck(boolean check) + { + assert check : "At most one task should be executing using this executor"; + } + + public T get(long timeout, TimeUnit timeUnit) throws InterruptedException, ExecutionException, TimeoutException + { + CompletableFuture lazyValue = cached.get(); + + // recompute was never called yet, return null. + if (lazyValue == null) + return null; + + return lazyValue.get(timeout, timeUnit); + } + + public String toString() + { + return "RecomputingSupplier{" + + "supplier=" + supplier + + ", cached=" + cached + + ", workInProgress=" + workInProgress + + ", executor=" + executor + + '}'; + } +} \ No newline at end of file diff --git a/test/distributed/org/apache/cassandra/distributed/test/GossipShutdownTest.java b/test/distributed/org/apache/cassandra/distributed/test/GossipShutdownTest.java index 9aebc2255e..b83689a804 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/GossipShutdownTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/GossipShutdownTest.java @@ -124,12 +124,5 @@ public class GossipShutdownTest extends TestBaseImpl { wasDead = true; } - - public void onRemove(InetAddress endpoint) {} - public void onRestart(InetAddress endpoint, EndpointState state) {} - public void onJoin(InetAddress endpoint, EndpointState epState) {} - public void beforeChange(InetAddress endpoint, EndpointState currentState, ApplicationState newStateKey, VersionedValue newValue) {} - public void onChange(InetAddress endpoint, ApplicationState state, VersionedValue value) {} - }; } diff --git a/test/distributed/org/apache/cassandra/distributed/test/NativeProtocolTest.java b/test/distributed/org/apache/cassandra/distributed/test/NativeProtocolTest.java index 0905b92f53..92751efe81 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/NativeProtocolTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/NativeProtocolTest.java @@ -18,7 +18,6 @@ package org.apache.cassandra.distributed.test; -import org.apache.cassandra.distributed.impl.RowUtil; import org.junit.Assert; import org.junit.Test; @@ -28,16 +27,17 @@ import com.datastax.driver.core.Session; import com.datastax.driver.core.SimpleStatement; import com.datastax.driver.core.Statement; import org.apache.cassandra.distributed.api.ICluster; +import org.apache.cassandra.distributed.impl.RowUtil; import static org.apache.cassandra.distributed.api.Feature.GOSSIP; import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL; import static org.apache.cassandra.distributed.api.Feature.NETWORK; -import static org.apache.cassandra.distributed.shared.AssertUtils.*; +import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows; +import static org.apache.cassandra.distributed.shared.AssertUtils.row; // TODO: this test should be removed after running in-jvm dtests is set up via the shared API repository public class NativeProtocolTest extends TestBaseImpl { - @Test public void withClientRequests() throws Throwable { diff --git a/test/distributed/org/apache/cassandra/distributed/test/ReprepareNewBehaviourTest.java b/test/distributed/org/apache/cassandra/distributed/test/ReprepareNewBehaviourTest.java new file mode 100644 index 0000000000..e961ad4788 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/ReprepareNewBehaviourTest.java @@ -0,0 +1,38 @@ +/* + * 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 org.junit.Test; + +public class ReprepareNewBehaviourTest extends ReprepareTestBase +{ + @Test + public void testReprepareNewBehaviour() throws Throwable + { + testReprepare(PrepareBehaviour::newBehaviour, + cfg(true, false), + cfg(false, false)); + } + + @Test + public void testReprepareTwoKeyspacesNewBehaviour() throws Throwable + { + testReprepareTwoKeyspaces(PrepareBehaviour::newBehaviour); + } +} \ No newline at end of file diff --git a/test/distributed/org/apache/cassandra/distributed/test/ReprepareTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/ReprepareTestBase.java new file mode 100644 index 0000000000..c270144276 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/ReprepareTestBase.java @@ -0,0 +1,281 @@ +/* + * 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.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.function.BiConsumer; + +import com.google.common.collect.Iterators; +import org.junit.Assert; + +import com.datastax.driver.core.Cluster; +import com.datastax.driver.core.Host; +import com.datastax.driver.core.HostDistance; +import com.datastax.driver.core.PreparedStatement; +import com.datastax.driver.core.Session; +import com.datastax.driver.core.Statement; +import com.datastax.driver.core.exceptions.DriverInternalError; +import com.datastax.driver.core.policies.LoadBalancingPolicy; +import net.bytebuddy.ByteBuddy; +import net.bytebuddy.dynamic.loading.ClassLoadingStrategy; +import net.bytebuddy.implementation.FixedValue; +import net.bytebuddy.implementation.MethodDelegation; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.cql3.statements.ParsedStatement; +import org.apache.cassandra.distributed.api.ICluster; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.exceptions.InvalidRequestException; +import org.apache.cassandra.service.ClientState; +import org.apache.cassandra.service.QueryState; +import org.apache.cassandra.transport.messages.ResultMessage; +import org.apache.cassandra.utils.FBUtilities; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; +import static org.apache.cassandra.distributed.shared.AssertUtils.fail; + +public class ReprepareTestBase extends TestBaseImpl +{ + protected static ReprepareTestConfiguration cfg(boolean withUse, boolean skipBrokenBehaviours) + { + return new ReprepareTestConfiguration(withUse, skipBrokenBehaviours); + } + + public void testReprepare(BiConsumer instanceInitializer, ReprepareTestConfiguration... configs) throws Throwable + { + try (ICluster c = init(builder().withNodes(2) + .withConfig(config -> config.with(GOSSIP, NETWORK, NATIVE_PROTOCOL)) + .withInstanceInitializer(instanceInitializer) + .start())) + { + ForceHostLoadBalancingPolicy lbp = new ForceHostLoadBalancingPolicy(); + c.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck));")); + + for (ReprepareTestConfiguration config : configs) + { + // 1 has old behaviour + for (int firstContact : new int[]{ 1, 2 }) + { + try (com.datastax.driver.core.Cluster cluster = com.datastax.driver.core.Cluster.builder() + .addContactPoint("127.0.0.1") + .addContactPoint("127.0.0.2") + .withLoadBalancingPolicy(lbp) + .build(); + Session session = cluster.connect()) + { + lbp.setPrimary(firstContact); + final PreparedStatement select = session.prepare(withKeyspace("SELECT * FROM %s.tbl")); + session.execute(select.bind()); + + c.stream().forEach((i) -> i.runOnInstance(QueryProcessor::clearPreparedStatementsCache)); + + lbp.setPrimary(firstContact == 1 ? 2 : 1); + + if (config.withUse) + session.execute(withKeyspace("USE %s")); + + // Re-preparing on the node + if (!config.skipBrokenBehaviours && firstContact == 1) + session.execute(select.bind()); + + c.stream().forEach((i) -> i.runOnInstance(QueryProcessor::clearPreparedStatementsCache)); + + lbp.setPrimary(firstContact); + + // Re-preparing on the node with old behaviour will break no matter where the statement was initially prepared + if (!config.skipBrokenBehaviours) + session.execute(select.bind()); + + c.stream().forEach((i) -> i.runOnInstance(QueryProcessor::clearPreparedStatementsCache)); + } + } + } + } + } + + public void testReprepareTwoKeyspaces(BiConsumer instanceInitializer) throws Throwable + { + try (ICluster c = init(builder().withNodes(2) + .withConfig(config -> config.with(GOSSIP, NETWORK, NATIVE_PROTOCOL)) + .withInstanceInitializer(instanceInitializer) + .start())) + { + c.schemaChange(withKeyspace("CREATE KEYSPACE %s2 WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 2};")); + c.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck));")); + + ForceHostLoadBalancingPolicy lbp = new ForceHostLoadBalancingPolicy(); + + for (int firstContact : new int[]{ 1, 2 }) + try (com.datastax.driver.core.Cluster cluster = com.datastax.driver.core.Cluster.builder() + .addContactPoint("127.0.0.1") + .addContactPoint("127.0.0.2") + .withLoadBalancingPolicy(lbp) + .build(); + Session session = cluster.connect()) + { + { + session.execute(withKeyspace("USE %s")); + c.stream().forEach((i) -> i.runOnInstance(QueryProcessor::clearPreparedStatementsCache)); + + lbp.setPrimary(firstContact); + final PreparedStatement select = session.prepare(withKeyspace("SELECT * FROM %s.tbl")); + session.execute(select.bind()); + + c.stream().forEach((i) -> i.runOnInstance(QueryProcessor::clearPreparedStatementsCache)); + + lbp.setPrimary(firstContact == 1 ? 2 : 1); + session.execute(withKeyspace("USE %s2")); + try + { + session.execute(select.bind()); + } + catch (DriverInternalError e) + { + Assert.assertTrue(e.getCause().getMessage().contains("can't execute it on")); + continue; + } + fail("Should have thrown"); + } + } + } + } + + protected static class ReprepareTestConfiguration + { + protected final boolean withUse; + protected final boolean skipBrokenBehaviours; + + protected ReprepareTestConfiguration(boolean withUse, boolean skipBrokenBehaviours) + { + this.withUse = withUse; + this.skipBrokenBehaviours = skipBrokenBehaviours; + } + } + + public static class PrepareBehaviour + { + protected static void setReleaseVersion(ClassLoader cl, String value) + { + new ByteBuddy().rebase(FBUtilities.class) + .method(named("getReleaseVersionString")) + .intercept(FixedValue.value(value)) + .make() + .load(cl, ClassLoadingStrategy.Default.INJECTION); + } + + static void newBehaviour(ClassLoader cl, int nodeNumber) + { + setReleaseVersion(cl, "3.0.19.63"); + } + + static void oldBehaviour(ClassLoader cl, int nodeNumber) + { + if (nodeNumber == 1) + { + new ByteBuddy().rebase(QueryProcessor.class) // note that we need to `rebase` when we use @SuperCall + .method(named("prepare").and(takesArguments(2))) + .intercept(MethodDelegation.to(PrepareBehaviour.class)) + .make() + .load(cl, ClassLoadingStrategy.Default.INJECTION); + setReleaseVersion(cl, "3.0.19.60"); + } + else + { + setReleaseVersion(cl, "3.0.19.63"); + } + } + + public static ResultMessage.Prepared prepare(String queryString, QueryState queryState) + { + ClientState clientState = queryState.getClientState(); + ResultMessage.Prepared existing = QueryProcessor.getStoredPreparedStatement(queryString, clientState.getRawKeyspace(), false); + if (existing != null) + return existing; + + ParsedStatement.Prepared prepared = QueryProcessor.getStatement(queryString, clientState); + int boundTerms = prepared.statement.getBoundTerms(); + if (boundTerms > FBUtilities.MAX_UNSIGNED_SHORT) + throw new InvalidRequestException(String.format("Too many markers(?). %d markers exceed the allowed maximum of %d", boundTerms, FBUtilities.MAX_UNSIGNED_SHORT)); + assert boundTerms == prepared.boundNames.size(); + + return QueryProcessor.storePreparedStatement(queryString, clientState.getRawKeyspace(), prepared, false); + } + + } + + protected static class ForceHostLoadBalancingPolicy implements LoadBalancingPolicy { + + protected final List hosts = new CopyOnWriteArrayList(); + protected int currentPrimary = 0; + + public void setPrimary(int idx) { + this.currentPrimary = idx - 1; // arrays are 0-based + } + + @Override + public void init(Cluster cluster, Collection hosts) { + this.hosts.addAll(hosts); + this.hosts.sort(Comparator.comparingInt(h -> h.getAddress().getAddress()[3])); + } + + @Override + public HostDistance distance(Host host) { + return HostDistance.LOCAL; + } + + @Override + public Iterator newQueryPlan(String loggedKeyspace, Statement statement) { + if (hosts.isEmpty()) return Collections.emptyIterator(); + return Iterators.singletonIterator(hosts.get(currentPrimary)); + } + + @Override + public void onAdd(Host host) { + onUp(host); + } + + @Override + public void onUp(Host host) { + hosts.add(host); + } + + @Override + public void onDown(Host host) { + // no-op + } + + @Override + public void onRemove(Host host) { + // no-op + } + + @Override + public void close() { + // no-op + } + } +} \ No newline at end of file diff --git a/test/distributed/org/apache/cassandra/distributed/test/ReprepareTestOldBehaviour.java b/test/distributed/org/apache/cassandra/distributed/test/ReprepareTestOldBehaviour.java new file mode 100644 index 0000000000..9900febaea --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/ReprepareTestOldBehaviour.java @@ -0,0 +1,129 @@ +/* + * 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 org.junit.Test; + +import com.datastax.driver.core.PreparedStatement; +import com.datastax.driver.core.Session; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.distributed.api.ICluster; +import org.apache.cassandra.distributed.api.IInvokableInstance; + +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; + +public class ReprepareTestOldBehaviour extends ReprepareTestBase +{ + @Test + public void testReprepareMixedVersion() throws Throwable + { + testReprepare(PrepareBehaviour::oldBehaviour, + cfg(true, true), + cfg(false, false)); + } + + @Test + public void testReprepareTwoKeyspacesMixedVersion() throws Throwable + { + testReprepareTwoKeyspaces(PrepareBehaviour::oldBehaviour); + } + + @Test + public void testReprepareUsingOldBehavior() throws Throwable + { + // fork of testReprepareMixedVersionWithoutReset, but makes sure oldBehavior has a clean state + try (ICluster c = init(builder().withNodes(2) + .withConfig(config -> config.with(GOSSIP, NETWORK, NATIVE_PROTOCOL)) + .withInstanceInitializer(PrepareBehaviour::oldBehaviour) + .start())) + { + ForceHostLoadBalancingPolicy lbp = new ForceHostLoadBalancingPolicy(); + c.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck));")); + + try (com.datastax.driver.core.Cluster cluster = com.datastax.driver.core.Cluster.builder() + .addContactPoint("127.0.0.1") + .addContactPoint("127.0.0.2") + .withLoadBalancingPolicy(lbp) + .build(); + Session session = cluster.connect()) + { + session.execute(withKeyspace("USE %s")); + + lbp.setPrimary(2); + final PreparedStatement select = session.prepare(withKeyspace("SELECT * FROM %s.tbl")); + session.execute(select.bind()); + + lbp.setPrimary(1); + session.execute(select.bind()); + } + } + } + + @Test + public void testReprepareMixedVersionWithoutReset() throws Throwable + { + try (ICluster c = init(builder().withNodes(2) + .withConfig(config -> config.with(GOSSIP, NETWORK, NATIVE_PROTOCOL)) + .withInstanceInitializer(PrepareBehaviour::oldBehaviour) + .start())) + { + ForceHostLoadBalancingPolicy lbp = new ForceHostLoadBalancingPolicy(); + c.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck));")); + + // 1 has old behaviour + for (int firstContact : new int[]{ 1, 2 }) + { + for (boolean withUse : new boolean[]{ true, false }) + { + for (boolean clearBetweenExecutions : new boolean[]{ true, false }) + { + try (com.datastax.driver.core.Cluster cluster = com.datastax.driver.core.Cluster.builder() + .addContactPoint("127.0.0.1") + .addContactPoint("127.0.0.2") + .withLoadBalancingPolicy(lbp) + .build(); + Session session = cluster.connect()) + { + if (withUse) + session.execute(withKeyspace("USE %s")); + + lbp.setPrimary(firstContact); + final PreparedStatement select = session.prepare(withKeyspace("SELECT * FROM %s.tbl")); + session.execute(select.bind()); + + if (clearBetweenExecutions) + c.get(2).runOnInstance(QueryProcessor::clearPreparedStatementsCache); + lbp.setPrimary(firstContact == 1 ? 2 : 1); + session.execute(select.bind()); + + if (clearBetweenExecutions) + c.get(2).runOnInstance(QueryProcessor::clearPreparedStatementsCache); + lbp.setPrimary(firstContact); + session.execute(select.bind()); + + c.get(2).runOnInstance(QueryProcessor::clearPreparedStatementsCache); + } + } + } + } + } + } +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java b/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java index e7adc8e614..753d6ff804 100644 --- a/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java +++ b/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java @@ -63,29 +63,29 @@ public class PstmtPersistenceTest extends CQLTester createTable("CREATE TABLE %s (pk int PRIMARY KEY, val text)"); List stmtIds = new ArrayList<>(); - // #0 - stmtIds.add(prepareStatement("SELECT * FROM %s WHERE keyspace_name = ?", SchemaConstants.SCHEMA_KEYSPACE_NAME, SchemaKeyspace.TABLES, clientState)); - // #1 - stmtIds.add(prepareStatement("SELECT * FROM %s WHERE pk = ?", clientState)); - // #2 - stmtIds.add(prepareStatement("SELECT * FROM %s WHERE key = ?", "foo", "bar", clientState)); + String statement0 = "SELECT * FROM %s WHERE keyspace_name = ?"; + String statement1 = "SELECT * FROM %s WHERE pk = ?"; + String statement2 = "SELECT * FROM %s WHERE key = ?"; + String statement3 = "SELECT * FROM %S WHERE key = ?"; + stmtIds.add(prepareStatement(statement0, SchemaConstants.SCHEMA_KEYSPACE_NAME, SchemaKeyspace.TABLES, clientState)); + stmtIds.add(prepareStatement(statement1, clientState)); + stmtIds.add(prepareStatement(statement2, "foo", "bar", clientState)); clientState.setKeyspace("foo"); - // #3 - stmtIds.add(prepareStatement("SELECT * FROM %s WHERE pk = ?", clientState)); - // #4 - stmtIds.add(prepareStatement("SELECT * FROM %S WHERE key = ?", "foo", "bar", clientState)); + + stmtIds.add(prepareStatement(statement1, clientState)); + stmtIds.add(prepareStatement(statement3, "foo", "bar", clientState)); assertEquals(5, stmtIds.size()); - assertEquals(5, QueryProcessor.preparedStatementsCount()); - - Assert.assertEquals(5, numberOfStatementsOnDisk()); + // statement1 will have two statements prepared because of `setKeyspace` usage + assertEquals(6, QueryProcessor.preparedStatementsCount()); + assertEquals(6, numberOfStatementsOnDisk()); QueryHandler handler = ClientState.getCQLQueryHandler(); validatePstmts(stmtIds, handler); // clear prepared statements cache QueryProcessor.clearPreparedStatements(true); - Assert.assertEquals(0, QueryProcessor.preparedStatementsCount()); + assertEquals(0, QueryProcessor.preparedStatementsCount()); for (MD5Digest stmtId : stmtIds) Assert.assertNull(handler.getPrepared(stmtId)); @@ -104,9 +104,11 @@ public class PstmtPersistenceTest extends CQLTester } // add anther prepared statement and sync it to table - prepareStatement("SELECT * FROM %s WHERE key = ?", "foo", "bar", clientState); - assertEquals(6, numberOfStatementsInMemory()); - assertEquals(6, numberOfStatementsOnDisk()); + prepareStatement(statement2, "foo", "bar", clientState); + + // statement1 will have two statements prepared because of `setKeyspace` usage + assertEquals(7, numberOfStatementsInMemory()); + assertEquals(7, numberOfStatementsOnDisk()); // drop a keyspace (prepared statements are removed - syncPreparedStatements() remove should the rows, too) execute("DROP KEYSPACE foo"); @@ -116,7 +118,6 @@ public class PstmtPersistenceTest extends CQLTester private void validatePstmts(List stmtIds, QueryHandler handler) { - assertEquals(5, QueryProcessor.preparedStatementsCount()); QueryOptions optionsStr = QueryOptions.forInternalCalls(Collections.singletonList(UTF8Type.instance.fromString("foobar"))); QueryOptions optionsInt = QueryOptions.forInternalCalls(Collections.singletonList(Int32Type.instance.decompose(42))); validatePstmt(handler, stmtIds.get(0), optionsStr); diff --git a/test/unit/org/apache/cassandra/metrics/CQLMetricsTest.java b/test/unit/org/apache/cassandra/metrics/CQLMetricsTest.java index 099a530dcb..72c839e604 100644 --- a/test/unit/org/apache/cassandra/metrics/CQLMetricsTest.java +++ b/test/unit/org/apache/cassandra/metrics/CQLMetricsTest.java @@ -63,9 +63,10 @@ public class CQLMetricsTest extends SchemaLoader @Test public void testPreparedStatementsCount() { - int n = (int) QueryProcessor.metrics.preparedStatementsCount.getValue(); + int n = QueryProcessor.metrics.preparedStatementsCount.getValue(); + session.execute("use junit"); session.prepare("SELECT * FROM junit.metricstest WHERE id = ?"); - assertEquals(n+1, (int) QueryProcessor.metrics.preparedStatementsCount.getValue()); + assertEquals(n+2, (int) QueryProcessor.metrics.preparedStatementsCount.getValue()); } @Test diff --git a/test/unit/org/apache/cassandra/utils/RecomputingSupplierTest.java b/test/unit/org/apache/cassandra/utils/RecomputingSupplierTest.java new file mode 100644 index 0000000000..c8292988b8 --- /dev/null +++ b/test/unit/org/apache/cassandra/utils/RecomputingSupplierTest.java @@ -0,0 +1,150 @@ +/* + * 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.utils; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.LockSupport; +import java.util.function.LongBinaryOperator; + +import org.junit.Assert; +import org.junit.Test; + +public class RecomputingSupplierTest +{ + @Test + public void recomputingSupplierTest() throws Throwable + { + ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(10); + ExecutorService testExecutor = Executors.newFixedThreadPool(10); + + AtomicReference thrown = new AtomicReference<>(); + CountDownLatch latch = new CountDownLatch(1); + + final AtomicLong counter = new AtomicLong(0); + final RecomputingSupplier supplier = new RecomputingSupplier<>(() -> { + try + { + long v = counter.incrementAndGet(); + LockSupport.parkNanos(1); + // Make sure that the value still hasn't changed + Assert.assertEquals(v, counter.get()); + return v; + } + catch (Throwable e) + { + thrown.set(e); + latch.countDown(); + throw new RuntimeException(e); + } + }, executor); + + for (int i = 0; i < 5; i++) + { + testExecutor.submit(() -> { + try + { + while (!Thread.interrupted() && !testExecutor.isShutdown()) + supplier.recompute(); + } + catch (Throwable e) + { + thrown.set(e); + latch.countDown(); + } + }); + } + + AtomicLong lastSeen = new AtomicLong(0); + for (int i = 0; i < 5; i++) + { + testExecutor.submit(() -> { + while (!Thread.interrupted() && !testExecutor.isShutdown()) + { + try + { + long seenBeforeGet = lastSeen.get(); + Long v = supplier.get(1, TimeUnit.SECONDS); + if (v != null) + { + lastSeen.accumulateAndGet(v, Math::max); + Assert.assertTrue(String.format("Seen %d out of order. Last seen value %d", v, seenBeforeGet), + v >= seenBeforeGet); + + } + + } + catch (Throwable e) + { + thrown.set(e); + latch.countDown(); + } + } + }); + } + + latch.await(10, TimeUnit.SECONDS); + + testExecutor.shutdown(); + testExecutor.awaitTermination(10, TimeUnit.SECONDS); + executor.shutdown(); + executor.awaitTermination(10, TimeUnit.SECONDS); + + if (thrown.get() != null) + throw new AssertionError(supplier.toString(), thrown.get()); + + Assert.assertTrue(counter.get() > 1); // We actually did some work + Assert.assertEquals(supplier.get(1, TimeUnit.SECONDS).longValue(), counter.get()); + } + + @Test + public void throwingSupplier() throws InterruptedException, TimeoutException + { + ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(1); + + + final RecomputingSupplier supplier = new RecomputingSupplier<>(() -> { + throw new RuntimeException(); + }, executor); + + supplier.recompute(); + + try + { + supplier.get(10, TimeUnit.SECONDS); + Assert.fail("Should have thrown"); + } + catch (ExecutionException t) + { + // ignore + } + finally + { + executor.shutdown(); + executor.awaitTermination(10, TimeUnit.SECONDS); + } + } +}