Merge branch 'cassandra-3.0' into cassandra-3.11

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
 =======
 

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)

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);
         }
     }
 

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);

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)

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<ColumnSpecification> boundNames;
+
+        @Nullable
         public final short[] partitionKeyBindIndexes;
 
-        protected Prepared(CQLStatement statement, List<ColumnSpecification> boundNames, short[] partitionKeyBindIndexes)
+        @Nullable
+        public final String keyspace;
+
+        protected Prepared(CQLStatement statement, List<ColumnSpecification> 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.<ColumnSpecification>emptyList(), null);
+            this(statement, Collections.<ColumnSpecification>emptyList(), null, null);
         }
     }
 

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);
         }
 
         /**

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<CassandraVersion> 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)
     {

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) {}
 }

src/java/org/apache/cassandra/repair/RepairSession.java

@@ -333,12 +333,6 @@ public class RepairSession extends AbstractFuture<RepairSessionResult> 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);

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)
     {

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());

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<T>
+{
+    private final Supplier<T> supplier;
+    private final AtomicReference<CompletableFuture<T>> cached = new AtomicReference<>(null);
+    private final AtomicBoolean workInProgress = new AtomicBoolean(false);
+    private final ExecutorService executor;
+
+    public RecomputingSupplier(Supplier<T> supplier, ExecutorService executor)
+    {
+        this.supplier = supplier;
+        this.executor = executor;
+    }
+
+    public void recompute()
+    {
+        CompletableFuture<T> 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<T> lazyValue = new CompletableFuture<>();
+        if (cached.compareAndSet(current, lazyValue))
+            executor.submit(() -> doWork(lazyValue));
+        else
+            executor.submit(this::recompute); // Lost CAS, resubmit
+    }
+
+    private void doWork(CompletableFuture<T> 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<T> 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 +
+               '}';
+    }
+}

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) {}
-
     };
 }

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
     {

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);
+    }
+}

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<ClassLoader, Integer> instanceInitializer, ReprepareTestConfiguration... configs) throws Throwable
+    {
+        try (ICluster<IInvokableInstance> 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<ClassLoader, Integer> instanceInitializer) throws Throwable
+    {
+        try (ICluster<IInvokableInstance> 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<Host> hosts = new CopyOnWriteArrayList<Host>();
+        protected int currentPrimary = 0;
+
+        public void setPrimary(int idx) {
+            this.currentPrimary = idx - 1; // arrays are 0-based
+        }
+
+        @Override
+        public void init(Cluster cluster, Collection<Host> 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<Host> 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
+        }
+    }
+}

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<IInvokableInstance> 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<IInvokableInstance> 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);
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

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<MD5Digest> 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<MD5Digest> 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);

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

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<Throwable> thrown = new AtomicReference<>();
+        CountDownLatch latch = new CountDownLatch(1);
+
+        final AtomicLong counter = new AtomicLong(0);
+        final RecomputingSupplier<Long> 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<Long> 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);
+        }
+    }
+}