diff --git a/src/java/org/apache/cassandra/cql3/CQLStatement.java b/src/java/org/apache/cassandra/cql3/CQLStatement.java index c34e27fb2e..e78f7332c3 100644 --- a/src/java/org/apache/cassandra/cql3/CQLStatement.java +++ b/src/java/org/apache/cassandra/cql3/CQLStatement.java @@ -110,4 +110,9 @@ public interface CQLStatement public abstract CQLStatement prepare(ClientState state); } + + public static interface SingleKeyspaceCqlStatement extends CQLStatement + { + public String keyspace(); + } } diff --git a/src/java/org/apache/cassandra/cql3/QueryProcessor.java b/src/java/org/apache/cassandra/cql3/QueryProcessor.java index dc1b7a74d1..42494e2b35 100644 --- a/src/java/org/apache/cassandra/cql3/QueryProcessor.java +++ b/src/java/org/apache/cassandra/cql3/QueryProcessor.java @@ -49,6 +49,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; @@ -64,6 +65,16 @@ public class QueryProcessor implements QueryHandler { public static final CassandraVersion CQL_VERSION = new CassandraVersion("3.4.5"); + /** + * 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); @@ -444,7 +455,30 @@ public class QueryProcessor implements QueryHandler 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)); - return storePreparedStatement(queryString, clientState.getRawKeyspace(), prepared); + if (statement instanceof CQLStatement.SingleKeyspaceCqlStatement) + { + // 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); + ResultMessage.Prepared oldBehavior = clientState.getRawKeyspace() != null ? storePreparedStatement(queryString, clientState.getRawKeyspace(), prepared) : 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); + } } private static MD5Digest computeId(String queryString, String keyspace) @@ -453,10 +487,11 @@ public class QueryProcessor implements QueryHandler return MD5Digest.compute(toHash); } - private static ResultMessage.Prepared getStoredPreparedStatement(String queryString, String keyspace) + @VisibleForTesting + public static ResultMessage.Prepared getStoredPreparedStatement(String queryString, String clientKeyspace) throws InvalidRequestException { - MD5Digest statementId = computeId(queryString, keyspace); + MD5Digest statementId = computeId(queryString, clientKeyspace); Prepared existing = preparedStatements.getIfPresent(statementId); if (existing == null) return null; @@ -469,7 +504,8 @@ public class QueryProcessor implements QueryHandler return new ResultMessage.Prepared(statementId, resultMetadata.getResultMetadataId(), preparedMetadata, resultMetadata); } - private static ResultMessage.Prepared storePreparedStatement(String queryString, String keyspace, Prepared prepared) + @VisibleForTesting + public static ResultMessage.Prepared storePreparedStatement(String queryString, String keyspace, Prepared prepared) throws InvalidRequestException { // Concatenate the current keyspace so we don't mix prepared statements between keyspace (#5352). @@ -609,6 +645,12 @@ public class QueryProcessor implements QueryHandler internalStatements.clear(); } + @VisibleForTesting + public static void clearPreparedStatementsCache() + { + preparedStatements.asMap().clear(); + } + private static class StatementInvalidatingListener extends SchemaChangeListener { private static void removeInvalidPreparedStatements(String ksName, String cfName) diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java index ca7d6d0ef2..e7b1577d3a 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java @@ -67,7 +67,7 @@ import static org.apache.cassandra.utils.Clock.Global.nanoTime; /* * Abstract parent class of individual modifications, i.e. INSERT, UPDATE and DELETE. */ -public abstract class ModificationStatement implements CQLStatement +public abstract class ModificationStatement implements CQLStatement.SingleKeyspaceCqlStatement { protected static final Logger logger = LoggerFactory.getLogger(ModificationStatement.class); @@ -197,6 +197,7 @@ public abstract class ModificationStatement implements CQLStatement public abstract void addUpdateForKey(PartitionUpdate.Builder updateBuilder, Slice slice, UpdateParameters params); + @Override public String keyspace() { return metadata.keyspace; diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java index 956e50b73b..d2643d1dbc 100644 --- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java @@ -91,7 +91,7 @@ import static org.apache.cassandra.utils.Clock.Global.nanoTime; * QueryHandler implementations, so before reducing their accessibility * due consideration should be given. */ -public class SelectStatement implements CQLStatement +public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement { private static final Logger logger = LoggerFactory.getLogger(SelectStatement.class); @@ -488,6 +488,7 @@ public class SelectStatement implements CQLStatement return process(partitions, options, selectors, nowInSec, getLimit(options)); } + @Override public String keyspace() { return table.keyspace; diff --git a/src/java/org/apache/cassandra/cql3/statements/UseStatement.java b/src/java/org/apache/cassandra/cql3/statements/UseStatement.java index ae6eeb0bb1..a0928e857c 100644 --- a/src/java/org/apache/cassandra/cql3/statements/UseStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/UseStatement.java @@ -78,4 +78,9 @@ public class UseStatement extends CQLStatement.Raw implements CQLStatement { return new AuditLogContext(AuditLogEntryType.USE_KEYSPACE, keyspace); } + + public String keyspace() + { + return keyspace; + } } diff --git a/src/java/org/apache/cassandra/cql3/statements/schema/AlterSchemaStatement.java b/src/java/org/apache/cassandra/cql3/statements/schema/AlterSchemaStatement.java index 161c9c4a93..124d04c354 100644 --- a/src/java/org/apache/cassandra/cql3/statements/schema/AlterSchemaStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/schema/AlterSchemaStatement.java @@ -35,7 +35,7 @@ import org.apache.cassandra.service.QueryState; import org.apache.cassandra.transport.Event.SchemaChange; import org.apache.cassandra.transport.messages.ResultMessage; -abstract class AlterSchemaStatement implements CQLStatement, SchemaTransformation +abstract public class AlterSchemaStatement implements CQLStatement.SingleKeyspaceCqlStatement, SchemaTransformation { protected final String keyspaceName; // name of the keyspace affected by the statement @@ -54,6 +54,12 @@ abstract class AlterSchemaStatement implements CQLStatement, SchemaTransformatio return execute(state, false); } + @Override + public String keyspace() + { + return keyspaceName; + } + public ResultMessage executeLocally(QueryState state, QueryOptions options) { return execute(state, true); diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index f0114c0281..dc1e3a185f 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -34,6 +34,7 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Iterables; import com.google.common.collect.ImmutableSet; +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; @@ -51,6 +52,7 @@ import org.apache.cassandra.utils.Pair; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import io.netty.util.concurrent.FastThreadLocal; import org.apache.cassandra.concurrent.DebuggableScheduledThreadPoolExecutor; import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor; import org.apache.cassandra.concurrent.Stage; @@ -61,8 +63,14 @@ import org.apache.cassandra.net.RequestCallback; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.utils.CassandraVersion; +import org.apache.cassandra.utils.ExecutorUtils; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.JVMStabilityInspector; +import org.apache.cassandra.utils.MBeanWrapper; +import org.apache.cassandra.utils.NoSpamLogger; +import org.apache.cassandra.utils.Pair; +import org.apache.cassandra.utils.RecomputingSupplier; import static org.apache.cassandra.config.CassandraRelevantProperties.GOSSIPER_QUARANTINE_DELAY; import static org.apache.cassandra.net.NoPayload.noPayload; @@ -335,7 +343,10 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean } } - Gossiper(boolean registerJmx) + private final RecomputingSupplier minVersionSupplier = new RecomputingSupplier<>(this::computeMinVersion, executor); + + @VisibleForTesting + public Gossiper(boolean registerJmx) { // half of QUARATINE_DELAY, to ensure justRemovedEndpoints has enough leeway to prevent re-gossip fatClientTimeout = (QUARANTINE_DELAY / 2); @@ -347,6 +358,31 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean { MBeanWrapper.instance.registerMBean(this, MBEAN_NAME); } + + subscribers.add(new IEndpointStateChangeSubscriber() + { + public void onJoin(InetAddressAndPort endpoint, EndpointState state) + { + maybeRecompute(state); + } + + public void onAlive(InetAddressAndPort endpoint, EndpointState state) + { + maybeRecompute(state); + } + + private void maybeRecompute(EndpointState state) + { + if (state.getApplicationState(ApplicationState.RELEASE_VERSION) != null) + minVersionSupplier.recompute(); + } + + public void onChange(InetAddressAndPort endpoint, ApplicationState state, VersionedValue value) + { + if (state == ApplicationState.RELEASE_VERSION) + minVersionSupplier.recompute(); + } + }); } public void setLastProcessedMessageAt(long timeInMillis) @@ -1716,6 +1752,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean maybeInitializeLocalState(generationNbr); EndpointState localState = endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort()); localState.addApplicationStates(preloadLocalStates); + minVersionSupplier.recompute(); //notify snitches that Gossiper is about to start DatabaseDescriptor.getEndpointSnitch().gossiperStarting(); @@ -2287,4 +2324,70 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean stop(); 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(InetAddressAndPort 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 (InetAddressAndPort 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; + } } diff --git a/src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java b/src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java index dc81650340..df046979d3 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(InetAddressAndPort endpoint, EndpointState epState); + default void onJoin(InetAddressAndPort endpoint, EndpointState epState) {} - public void beforeChange(InetAddressAndPort endpoint, EndpointState currentState, ApplicationState newStateKey, VersionedValue newValue); + default void beforeChange(InetAddressAndPort endpoint, EndpointState currentState, ApplicationState newStateKey, VersionedValue newValue) {} - public void onChange(InetAddressAndPort endpoint, ApplicationState state, VersionedValue value); + default void onChange(InetAddressAndPort endpoint, ApplicationState state, VersionedValue value) {} - public void onAlive(InetAddressAndPort endpoint, EndpointState state); + default void onAlive(InetAddressAndPort endpoint, EndpointState state) {} - public void onDead(InetAddressAndPort endpoint, EndpointState state); + default void onDead(InetAddressAndPort endpoint, EndpointState state) {} - public void onRemove(InetAddressAndPort endpoint); + default void onRemove(InetAddressAndPort 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(InetAddressAndPort endpoint, EndpointState state); + default void onRestart(InetAddressAndPort endpoint, EndpointState state) {} } diff --git a/src/java/org/apache/cassandra/net/StartupClusterConnectivityChecker.java b/src/java/org/apache/cassandra/net/StartupClusterConnectivityChecker.java index 073db339f2..cd607a4780 100644 --- a/src/java/org/apache/cassandra/net/StartupClusterConnectivityChecker.java +++ b/src/java/org/apache/cassandra/net/StartupClusterConnectivityChecker.java @@ -222,18 +222,6 @@ public class StartupClusterConnectivityChecker this.getDatacenter = getDatacenter; } - public void onJoin(InetAddressAndPort endpoint, EndpointState epState) - { - } - - public void beforeChange(InetAddressAndPort endpoint, EndpointState currentState, ApplicationState newStateKey, VersionedValue newValue) - { - } - - public void onChange(InetAddressAndPort endpoint, ApplicationState state, VersionedValue value) - { - } - public void onAlive(InetAddressAndPort endpoint, EndpointState state) { if (livePeers.add(endpoint) && acks.incrementAndCheck(endpoint)) @@ -243,18 +231,6 @@ public class StartupClusterConnectivityChecker dcToRemainingPeers.get(datacenter).countDown(); } } - - public void onDead(InetAddressAndPort endpoint, EndpointState state) - { - } - - public void onRemove(InetAddressAndPort endpoint) - { - } - - public void onRestart(InetAddressAndPort endpoint, EndpointState state) - { - } } private static final class AckMap diff --git a/src/java/org/apache/cassandra/repair/RepairSession.java b/src/java/org/apache/cassandra/repair/RepairSession.java index dad1673449..4e5daf52cd 100644 --- a/src/java/org/apache/cassandra/repair/RepairSession.java +++ b/src/java/org/apache/cassandra/repair/RepairSession.java @@ -360,12 +360,6 @@ public class RepairSession extends AbstractFuture implement terminate(); } - public void onJoin(InetAddressAndPort endpoint, EndpointState epState) {} - public void beforeChange(InetAddressAndPort endpoint, EndpointState currentState, ApplicationState newStateKey, VersionedValue newValue) {} - public void onChange(InetAddressAndPort endpoint, ApplicationState state, VersionedValue value) {} - public void onAlive(InetAddressAndPort endpoint, EndpointState state) {} - public void onDead(InetAddressAndPort endpoint, EndpointState state) {} - public void onRemove(InetAddressAndPort 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 113aead3c6..ebda3cdb3e 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(InetAddressAndPort endpoint, EndpointState currentState, ApplicationState newStateKey, VersionedValue newValue) {} - - public void onAlive(InetAddressAndPort endpoint, EndpointState state) {} - - public void onDead(InetAddressAndPort endpoint, EndpointState state) {} - - public void onRestart(InetAddressAndPort endpoint, EndpointState state) {} public void onRemove(InetAddressAndPort endpoint) { diff --git a/src/java/org/apache/cassandra/streaming/StreamSession.java b/src/java/org/apache/cassandra/streaming/StreamSession.java index bbbe79d196..7d6e6da9ac 100644 --- a/src/java/org/apache/cassandra/streaming/StreamSession.java +++ b/src/java/org/apache/cassandra/streaming/StreamSession.java @@ -911,12 +911,6 @@ public class StreamSession implements IEndpointStateChangeSubscriber maybeCompleted(); } - public void onJoin(InetAddressAndPort endpoint, EndpointState epState) {} - public void beforeChange(InetAddressAndPort endpoint, EndpointState currentState, ApplicationState newStateKey, VersionedValue newValue) {} - public void onChange(InetAddressAndPort endpoint, ApplicationState state, VersionedValue value) {} - public void onAlive(InetAddressAndPort endpoint, EndpointState state) {} - public void onDead(InetAddressAndPort endpoint, EndpointState state) {} - public void onRemove(InetAddressAndPort endpoint) { logger.error("[Stream #{}] Session failed because remote peer {} has left.", planId(), peer.toString()); 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 02e807532b..ad21c21694 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/GossipShutdownTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/GossipShutdownTest.java @@ -125,12 +125,5 @@ public class GossipShutdownTest extends TestBaseImpl { wasDead = true; } - - public void onRemove(InetAddressAndPort endpoint) {} - public void onRestart(InetAddressAndPort endpoint, EndpointState state) {} - public void onJoin(InetAddressAndPort endpoint, EndpointState epState) {} - public void beforeChange(InetAddressAndPort endpoint, EndpointState currentState, ApplicationState newStateKey, VersionedValue newValue) {} - public void onChange(InetAddressAndPort 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..60fe5b6e36 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; @@ -27,17 +26,26 @@ import com.datastax.driver.core.ResultSet; import com.datastax.driver.core.Session; import com.datastax.driver.core.SimpleStatement; 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.MethodDelegation; +import org.apache.cassandra.cql3.CQLStatement; +import org.apache.cassandra.cql3.QueryHandler; +import org.apache.cassandra.cql3.QueryProcessor; 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..02af6d5078 --- /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.concurrent.TimeUnit; +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.CQLStatement; +import org.apache.cassandra.cql3.QueryHandler; +import org.apache.cassandra.cql3.QueryProcessor; +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.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, ClientState clientState) + { + ResultMessage.Prepared existing = QueryProcessor.getStoredPreparedStatement(queryString, clientState.getRawKeyspace()); + if (existing != null) + return existing; + + CQLStatement statement = QueryProcessor.getStatement(queryString, clientState); + QueryHandler.Prepared prepared = new QueryHandler.Prepared(statement, queryString); + + int boundTerms = statement.getBindVariables().size(); + 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)); + + return QueryProcessor.storePreparedStatement(queryString, clientState.getRawKeyspace(), prepared); + } + } + + 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/PreparedStatementsTest.java b/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java index 5b3ae3e8a6..ffd7e25c21 100644 --- a/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java +++ b/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java @@ -231,9 +231,9 @@ public class PreparedStatementsTest extends CQLTester newSession.execute("USE " + keyspace()); preparedInsert = newSession.prepare(insertCQL); preparedSelect = newSession.prepare(selectCQL); - session.execute(preparedInsert.bind(1, 1, "value")); + newSession.execute(preparedInsert.bind(1, 1, "value")); - assertEquals(1, session.execute(preparedSelect.bind(1)).all().size()); + assertEquals(1, newSession.execute(preparedSelect.bind(1)).all().size()); } } } diff --git a/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java b/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java index da9fc2a551..a649532292 100644 --- a/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java +++ b/test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java @@ -65,22 +65,22 @@ 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()); - - 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); @@ -106,9 +106,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"); @@ -118,7 +120,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); @@ -188,6 +189,7 @@ public class PstmtPersistenceTest extends CQLTester private MD5Digest prepareStatement(String stmt, String keyspace, String table, ClientState clientState) { + System.out.println(stmt + String.format(stmt, keyspace + "." + table)); return QueryProcessor.prepare(String.format(stmt, keyspace + "." + table), clientState).statementId; } } diff --git a/test/unit/org/apache/cassandra/metrics/CQLMetricsTest.java b/test/unit/org/apache/cassandra/metrics/CQLMetricsTest.java index fbd825e8ae..3971b9f1e1 100644 --- a/test/unit/org/apache/cassandra/metrics/CQLMetricsTest.java +++ b/test/unit/org/apache/cassandra/metrics/CQLMetricsTest.java @@ -60,9 +60,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); + } + } +}