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

Merge branch 'cassandra-4.0' into trunk

This commit is contained in:
Alex Petrov 2021-10-03 23:24:22 +02:00
parent 0945002457 3660a58ab4
commit c3c2c7efec
22 changed files with 933 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/java/org/apache/cassandra/cql3/CQLStatement.java
View File

@ -110,4 +110,9 @@ public interface CQLStatement
public abstract CQLStatement prepare(ClientState state);
}
public static interface SingleKeyspaceCqlStatement extends CQLStatement
{
public String keyspace();
}
}

50
src/java/org/apache/cassandra/cql3/QueryProcessor.java
View File

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

3
src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
View File

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

3
src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
View File

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

5
src/java/org/apache/cassandra/cql3/statements/UseStatement.java
View File

@ -78,4 +78,9 @@ public class UseStatement extends CQLStatement.Raw implements CQLStatement
{
return new AuditLogContext(AuditLogEntryType.USE_KEYSPACE, keyspace);
}
public String keyspace()
{
return keyspace;
}
}

8
src/java/org/apache/cassandra/cql3/statements/schema/AlterSchemaStatement.java
View File

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

105
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -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<CassandraVersion> 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;
}
}

14
src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java
View File

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

24
src/java/org/apache/cassandra/net/StartupClusterConnectivityChecker.java
View File

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

6
src/java/org/apache/cassandra/repair/RepairSession.java
View File

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

8
src/java/org/apache/cassandra/service/LoadBroadcaster.java
View File

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

6
src/java/org/apache/cassandra/streaming/StreamSession.java
View File

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

122
src/java/org/apache/cassandra/utils/RecomputingSupplier.java Normal file
View File

@ -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 +
'}';
}
}

7
test/distributed/org/apache/cassandra/distributed/test/GossipShutdownTest.java
View File

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

14
test/distributed/org/apache/cassandra/distributed/test/NativeProtocolTest.java
View File

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

38
test/distributed/org/apache/cassandra/distributed/test/ReprepareNewBehaviourTest.java Normal file
View File

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

281
test/distributed/org/apache/cassandra/distributed/test/ReprepareTestBase.java Normal file
View File

@ -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<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, 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<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
}
}
}

129
test/distributed/org/apache/cassandra/distributed/test/ReprepareTestOldBehaviour.java Normal file
View File

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

4
test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
View File

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

36
test/unit/org/apache/cassandra/cql3/PstmtPersistenceTest.java
View File

@ -65,22 +65,22 @@ 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());
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<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);
@ -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;
}
}

5
test/unit/org/apache/cassandra/metrics/CQLMetricsTest.java
View File

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

150
test/unit/org/apache/cassandra/utils/RecomputingSupplierTest.java Normal file
View File

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