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

Merge branch 'cassandra-3.0' into cassandra-3.11

This commit is contained in:
Dinesh A. Joshi 2019-10-08 14:07:10 -07:00
parent 3040965668 93815db985
commit 4675b3fd33
21 changed files with 743 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -43,6 +43,7 @@ Merged from 2.2:
* Fixing invalid CQL in security documentation (CASSANDRA-15020)
* Multi-version in-JVM dtests (CASSANDRA-14937)
* Allow instance class loaders to be garbage collected for inJVM dtest (CASSANDRA-15170)
* Add support for network topology and query tracing for inJVM dtest (CASSANDRA-15319)
3.11.4

2
build.xml
View File

@ -1264,7 +1264,7 @@
algorithm to limit the metaspace size and clean up SoftReferences
more aggressively rather than waiting. See CASSANDRA-14922 for more details.
-->
<jvmarg value="-XX:MaxMetaspaceSize=256M" />
<jvmarg value="-XX:MaxMetaspaceSize=384M" />
<jvmarg value="-XX:SoftRefLRUPolicyMSPerMB=0" />
<jvmarg value="-Dcassandra.memtable_row_overhead_computation_step=100"/>
<jvmarg value="-Dcassandra.test.use_prepared=${cassandra.test.use_prepared}"/>

2
ide/idea/workspace.xml
View File

@ -167,7 +167,7 @@
<option name="MAIN_CLASS_NAME" value="" />
<option name="METHOD_NAME" value="" />
<option name="TEST_OBJECT" value="class" />
<option name="VM_PARAMETERS" value="-Dcassandra.config=file://$PROJECT_DIR$/test/conf/cassandra.yaml -Dlogback.configurationFile=file://$PROJECT_DIR$/test/conf/logback-test.xml -Dcassandra.logdir=$PROJECT_DIR$/build/test/logs -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin -ea -XX:MaxMetaspaceSize=256M -XX:SoftRefLRUPolicyMSPerMB=0 -Dcassandra.strict.runtime.checks=true" />
<option name="VM_PARAMETERS" value="-Dcassandra.config=file://$PROJECT_DIR$/test/conf/cassandra.yaml -Dlogback.configurationFile=file://$PROJECT_DIR$/test/conf/logback-test.xml -Dcassandra.logdir=$PROJECT_DIR$/build/test/logs -Djava.library.path=$PROJECT_DIR$/lib/sigar-bin -ea -XX:MaxMetaspaceSize=384M -XX:SoftRefLRUPolicyMSPerMB=0 -Dcassandra.strict.runtime.checks=true" />
<option name="PARAMETERS" value="" />
<option name="WORKING_DIRECTORY" value="" />
<option name="ENV_VARIABLES" />

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

@ -1039,7 +1039,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
}
// if we don't have system_traces keyspace at this point, then create it manually
maybeAddOrUpdateKeyspace(TraceKeyspace.metadata());
ensureTraceKeyspace();
maybeAddOrUpdateKeyspace(SystemDistributedKeyspace.metadata());
if (!isSurveyMode)
@ -1070,6 +1070,12 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
}
}
@VisibleForTesting
public void ensureTraceKeyspace()
{
maybeAddOrUpdateKeyspace(TraceKeyspace.metadata());
}
public static boolean isReplacingSameAddress()
{
InetAddress replaceAddress = DatabaseDescriptor.getReplaceAddress();

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

@ -45,9 +45,14 @@ public class Cluster extends AbstractCluster<IInvokableInstance> implements IClu
return new Wrapper(generation, version, config);
}
public static Builder<IInvokableInstance, Cluster> build()
{
return new Builder<>(Cluster::new);
}
public static Builder<IInvokableInstance, Cluster> build(int nodeCount)
{
return new Builder<>(nodeCount, Cluster::new);
return build().withNodes(nodeCount);
}
public static Cluster create(int nodeCount, Consumer<InstanceConfig> configUpdater) throws IOException

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

@ -46,9 +46,14 @@ public class UpgradeableCluster extends AbstractCluster<IUpgradeableInstance> im
return new Wrapper(generation, version, config);
}
public static Builder<IUpgradeableInstance, UpgradeableCluster> build()
{
return new Builder<>(UpgradeableCluster::new);
}
public static Builder<IUpgradeableInstance, UpgradeableCluster> build(int nodeCount)
{
return new Builder<>(nodeCount, UpgradeableCluster::new);
return build().withNodes(nodeCount);
}
public static UpgradeableCluster create(int nodeCount) throws Throwable

2
test/distributed/org/apache/cassandra/distributed/api/ICluster.java
View File

@ -29,6 +29,8 @@ public interface ICluster
IInstance get(InetAddressAndPort endpoint);
int size();
Stream<? extends IInstance> stream();
Stream<? extends IInstance> stream(String dcName);
Stream<? extends IInstance> stream(String dcName, String rackName);
IMessageFilters filters();
}

5
test/distributed/org/apache/cassandra/distributed/api/ICoordinator.java
View File

@ -19,6 +19,8 @@
package org.apache.cassandra.distributed.api;
import java.util.Iterator;
import java.util.UUID;
import java.util.concurrent.Future;
// The cross-version API requires that a Coordinator can be constructed without any constructor arguments
public interface ICoordinator
@ -28,4 +30,7 @@ public interface ICoordinator
Object[][] execute(String query, Enum<?> consistencyLevel, Object... boundValues);
Iterator<Object[]> executeWithPaging(String query, Enum<?> consistencyLevel, int pageSize, Object... boundValues);
Future<Object[][]> asyncExecuteWithTracing(UUID sessionId, String query, Enum<?> consistencyLevel, Object... boundValues);
Object[][] executeWithTracing(UUID sessionId, String query, Enum<?> consistencyLevel, Object... boundValues);
}

14
test/distributed/org/apache/cassandra/distributed/api/IInstanceConfig.java
View File

@ -18,8 +18,11 @@
package org.apache.cassandra.distributed.api;
import org.apache.cassandra.distributed.impl.NetworkTopology;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.utils.Pair;
import java.util.Map;
import java.util.UUID;
public interface IInstanceConfig
@ -27,6 +30,17 @@ public interface IInstanceConfig
int num();
UUID hostId();
InetAddressAndPort broadcastAddressAndPort();
NetworkTopology networkTopology();
default public String localRack()
{
return networkTopology().localRack(broadcastAddressAndPort());
}
default public String localDatacenter()
{
return networkTopology().localDC(broadcastAddressAndPort());
}
/**
* write the specified parameters to the Config object; we do not specify Config as the type to support a Config

153
test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
View File

@ -35,6 +35,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import com.google.common.collect.Sets;
@ -57,6 +58,7 @@ import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
/**
@ -193,13 +195,15 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
}
}
protected AbstractCluster(File root, Versions.Version version, List<InstanceConfig> configs, ClassLoader sharedClassLoader)
protected AbstractCluster(File root, Versions.Version version, List<InstanceConfig> configs,
ClassLoader sharedClassLoader)
{
this.root = root;
this.sharedClassLoader = sharedClassLoader;
this.instances = new ArrayList<>();
this.instanceMap = new HashMap<>();
int generation = AbstractCluster.generation.incrementAndGet();
for (InstanceConfig config : configs)
{
I instance = newInstanceWrapper(generation, version, config);
@ -231,7 +235,20 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
{
return instances.size();
}
public Stream<I> stream() { return instances.stream(); }
public Stream<I> stream(String dcName)
{
return instances.stream().filter(i -> i.config().localDatacenter().equals(dcName));
}
public Stream<I> stream(String dcName, String rackName)
{
return instances.stream().filter(i -> i.config().localDatacenter().equals(dcName) &&
i.config().localRack().equals(rackName));
}
public void forEach(IIsolatedExecutor.SerializableRunnable runnable) { forEach(i -> i.sync(runnable)); }
public void forEach(Consumer<? super I> consumer) { instances.forEach(consumer); }
public void parallelForEach(IIsolatedExecutor.SerializableConsumer<? super I> consumer, long timeout, TimeUnit units)
@ -309,7 +326,7 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
private void signal()
{
if (schemaHasChanged && 1 == instances.stream().map(IInstance::schemaVersion).distinct().count())
if (schemaHasChanged && 1 == instances.stream().filter(i -> !i.isShutdown()).map(IInstance::schemaVersion).distinct().count())
agreement.signalAll();
}
@ -353,15 +370,16 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
public static class Builder<I extends IInstance, C extends AbstractCluster<I>>
{
private final int nodeCount;
private final Factory<I, C> factory;
private int nodeCount;
private int subnet;
private Map<Integer, Pair<String,String>> nodeIdTopology;
private File root;
private Versions.Version version;
private Consumer<InstanceConfig> configUpdater;
public Builder(int nodeCount, Factory<I, C> factory)
public Builder(Factory<I, C> factory)
{
this.nodeCount = nodeCount;
this.factory = factory;
}
@ -371,6 +389,97 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
return this;
}
public Builder<I, C> withNodes(int nodeCount) {
this.nodeCount = nodeCount;
return this;
}
public Builder<I, C> withDCs(int dcCount)
{
return withRacks(dcCount, 1);
}
public Builder<I, C> withRacks(int dcCount, int racksPerDC)
{
if (nodeCount == 0)
throw new IllegalStateException("Node count will be calculated. Do not supply total node count in the builder");
int totalRacks = dcCount * racksPerDC;
int nodesPerRack = (nodeCount + totalRacks - 1) / totalRacks; // round up to next integer
return withRacks(dcCount, racksPerDC, nodesPerRack);
}
public Builder<I, C> withRacks(int dcCount, int racksPerDC, int nodesPerRack)
{
if (nodeIdTopology != null)
throw new IllegalStateException("Network topology already created. Call withDCs/withRacks once or before withDC/withRack calls");
nodeIdTopology = new HashMap<>();
int nodeId = 1;
for (int dc = 1; dc <= dcCount; dc++)
{
for (int rack = 1; rack <= racksPerDC; rack++)
{
for (int rackNodeIdx = 0; rackNodeIdx < nodesPerRack; rackNodeIdx++)
nodeIdTopology.put(nodeId++, Pair.create(dcName(dc), rackName(rack)));
}
}
// adjust the node count to match the allocatation
final int adjustedNodeCount = dcCount * racksPerDC * nodesPerRack;
if (adjustedNodeCount != nodeCount)
{
assert adjustedNodeCount > nodeCount : "withRacks should only ever increase the node count";
logger.info("Network topology of {} DCs with {} racks per DC and {} nodes per rack required increasing total nodes to {}",
dcCount, racksPerDC, nodesPerRack, adjustedNodeCount);
nodeCount = adjustedNodeCount;
}
return this;
}
public Builder<I, C> withDC(String dcName, int nodeCount)
{
return withRack(dcName, rackName(1), nodeCount);
}
public Builder<I, C> withRack(String dcName, String rackName, int nodesInRack)
{
if (nodeIdTopology == null)
{
if (nodeCount > 0)
throw new IllegalStateException("Node count must not be explicitly set, or allocated using withDCs/withRacks");
nodeIdTopology = new HashMap<>();
}
for (int nodeId = nodeCount + 1; nodeId <= nodeCount + nodesInRack; nodeId++)
nodeIdTopology.put(nodeId, Pair.create(dcName, rackName));
nodeCount += nodesInRack;
return this;
}
// Map of node ids to dc and rack - must be contiguous with an entry nodeId 1 to nodeCount
public Builder<I, C> withNodeIdTopology(Map<Integer,Pair<String,String>> nodeIdTopology)
{
if (nodeIdTopology.isEmpty())
throw new IllegalStateException("Topology is empty. It must have an entry for every nodeId.");
IntStream.rangeClosed(1, nodeIdTopology.size()).forEach(nodeId -> {
if (nodeIdTopology.get(nodeId) == null)
throw new IllegalStateException("Topology is missing entry for nodeId " + nodeId);
});
if (nodeCount != nodeIdTopology.size())
{
nodeCount = nodeIdTopology.size();
logger.info("Adjusting node count to {} for supplied network topology", nodeCount);
}
this.nodeIdTopology = new HashMap<>(nodeIdTopology);
return this;
}
public Builder<I, C> withRoot(File root)
{
this.root = root;
@ -396,9 +505,18 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
if (root == null)
root = Files.createTempDirectory("dtests").toFile();
if (version == null)
version = Versions.CURRENT;
if (nodeCount <= 0)
throw new IllegalStateException("Cluster must have at least one node");
if (nodeIdTopology == null)
nodeIdTopology = IntStream.rangeClosed(1, nodeCount).boxed()
.collect(Collectors.toMap(nodeId -> nodeId,
nodeId -> Pair.create(dcName(0), rackName(0))));
root.mkdirs();
setupLogging(root);
@ -406,17 +524,23 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
List<InstanceConfig> configs = new ArrayList<>();
long token = Long.MIN_VALUE + 1, increment = 2 * (Long.MAX_VALUE / nodeCount);
for (int i = 0; i < nodeCount; ++i)
String ipPrefix = "127.0." + subnet + ".";
NetworkTopology networkTopology = NetworkTopology.build(ipPrefix, 7012, nodeIdTopology);
for (int i = 0 ; i < nodeCount ; ++i)
{
InstanceConfig config = InstanceConfig.generate(i + 1, subnet, root, String.valueOf(token));
int nodeNum = i + 1;
String ipAddress = ipPrefix + nodeNum;
InstanceConfig config = InstanceConfig.generate(i + 1, ipAddress, networkTopology, root, String.valueOf(token));
if (configUpdater != null)
configUpdater.accept(config);
configs.add(config);
token += increment;
}
C cluster = factory.newCluster(root, version, configs, sharedClassLoader);
return cluster;
return factory.newCluster(root, version, configs, sharedClassLoader);
}
public C start() throws IOException
@ -427,6 +551,15 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
}
}
static String dcName(int index)
{
return "datacenter" + index;
}
static String rackName(int index)
{
return "rack" + index;
}
private static void setupLogging(File root)
{
@ -434,11 +567,13 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster,
{
String testConfPath = "test/conf/logback-dtest.xml";
Path logConfPath = Paths.get(root.getPath(), "/logback-dtest.xml");
if (!logConfPath.toFile().exists())
{
Files.copy(new File(testConfPath).toPath(),
logConfPath);
}
System.setProperty("logback.configurationFile", "file://" + logConfPath);
}
catch (IOException e)

69
test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java
View File

@ -21,8 +21,11 @@ package org.apache.cassandra.distributed.impl;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Future;
import org.apache.cassandra.cql3.CQLStatement;
import org.apache.cassandra.cql3.QueryOptions;
@ -35,6 +38,7 @@ import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.service.pager.QueryPager;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
@ -50,37 +54,56 @@ public class Coordinator implements ICoordinator
@Override
public Object[][] execute(String query, Enum<?> consistencyLevelOrigin, Object... boundValues)
{
return instance.sync(() -> {
ConsistencyLevel consistencyLevel = ConsistencyLevel.valueOf(consistencyLevelOrigin.name());
ClientState clientState = makeFakeClientState();
CQLStatement prepared = QueryProcessor.getStatement(query, clientState).statement;
List<ByteBuffer> boundBBValues = new ArrayList<>();
for (Object boundValue : boundValues)
{
boundBBValues.add(ByteBufferUtil.objectToBytes(boundValue));
}
return instance.sync(() -> executeInternal(query, consistencyLevelOrigin, boundValues)).call();
}
ResultMessage res = prepared.execute(QueryState.forInternalCalls(),
QueryOptions.create(consistencyLevel,
boundBBValues,
false,
Integer.MAX_VALUE,
null,
null,
ProtocolVersion.CURRENT),
System.nanoTime());
if (res != null && res.kind == ResultMessage.Kind.ROWS)
public Future<Object[][]> asyncExecuteWithTracing(UUID sessionId, String query, Enum<?> consistencyLevelOrigin, Object... boundValues)
{
return instance.async(() -> {
try
{
return RowUtil.toObjects((ResultMessage.Rows) res);
Tracing.instance.newSession(sessionId, Collections.emptyMap());
return executeInternal(query, consistencyLevelOrigin, boundValues);
}
else
finally
{
return new Object[][]{};
Tracing.instance.stopSession();
}
}).call();
}
private Object[][] executeInternal(String query, Enum<?> consistencyLevelOrigin, Object[] boundValues)
{
ConsistencyLevel consistencyLevel = ConsistencyLevel.valueOf(consistencyLevelOrigin.name());
ClientState clientState = makeFakeClientState();
CQLStatement prepared = QueryProcessor.getStatement(query, clientState).statement;
List<ByteBuffer> boundBBValues = new ArrayList<>();
for (Object boundValue : boundValues)
{
boundBBValues.add(ByteBufferUtil.objectToBytes(boundValue));
}
ResultMessage res = prepared.execute(QueryState.forInternalCalls(),
QueryOptions.create(consistencyLevel,
boundBBValues,
false,
Integer.MAX_VALUE,
null,
null,
ProtocolVersion.CURRENT),
System.nanoTime());
if (res != null && res.kind == ResultMessage.Kind.ROWS)
return RowUtil.toObjects((ResultMessage.Rows) res);
else
return new Object[][]{};
}
public Object[][] executeWithTracing(UUID sessionId, String query, Enum<?> consistencyLevelOrigin, Object... boundValues)
{
return IsolatedExecutor.waitOn(asyncExecuteWithTracing(sessionId, query, consistencyLevelOrigin, boundValues));
}
@Override
public Iterator<Object[]> executeWithPaging(String query, Enum<?> consistencyLevelOrigin, int pageSize, Object... boundValues)
{

61
test/distributed/org/apache/cassandra/distributed/impl/DistributedTestSnitch.java Normal file
View File

@ -0,0 +1,61 @@
/*
* 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.impl;
import java.net.InetAddress;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.locator.AbstractNetworkTopologySnitch;
import org.apache.cassandra.locator.InetAddressAndPort;
public class DistributedTestSnitch extends AbstractNetworkTopologySnitch
{
private static NetworkTopology mapping = null;
public String getRack(InetAddress endpoint)
{
assert mapping != null : "network topology must be assigned before using snitch";
int storage_port = Config.getOverrideLoadConfig().get().storage_port;
return mapping.localRack(InetAddressAndPort.getByAddressOverrideDefaults(endpoint, storage_port));
}
public String getRack(InetAddressAndPort endpoint)
{
assert mapping != null : "network topology must be assigned before using snitch";
return mapping.localRack(endpoint);
}
public String getDatacenter(InetAddress endpoint)
{
assert mapping != null : "network topology must be assigned before using snitch";
int storage_port = Config.getOverrideLoadConfig().get().storage_port;
return mapping.localDC(InetAddressAndPort.getByAddressOverrideDefaults(endpoint, storage_port));
}
public String getDatacenter(InetAddressAndPort endpoint)
{
assert mapping != null : "network topology must be assigned before using snitch";
return mapping.localDC(endpoint);
}
static void assign(NetworkTopology newMapping)
{
mapping = new NetworkTopology(newMapping);
}
}

32
test/distributed/org/apache/cassandra/distributed/impl/Instance.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.distributed.impl;
import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@ -72,7 +73,6 @@ import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.IMessageSink;
import org.apache.cassandra.net.MessageDeliveryTask;
import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.net.MessageOut;
import org.apache.cassandra.net.MessagingService;
@ -83,10 +83,13 @@ import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.StreamCoordinator;
import org.apache.cassandra.streaming.StreamSession;
import org.apache.cassandra.tracing.TraceState;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.NanoTimeToCurrentTimeMillis;
import org.apache.cassandra.utils.Throwables;
import org.apache.cassandra.utils.UUIDGen;
import org.apache.cassandra.utils.concurrent.Ref;
import org.apache.cassandra.utils.memory.BufferPool;
@ -246,6 +249,28 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
InetAddressAndPort toFull = lookupAddressAndPort.apply(to);
int version = MessagingService.instance().getVersion(to);
// Tracing logic - similar to org.apache.cassandra.net.OutboundTcpConnection.writeConnected
byte[] sessionBytes = (byte[]) messageOut.parameters.get(Tracing.TRACE_HEADER);
if (sessionBytes != null)
{
UUID sessionId = UUIDGen.getUUID(ByteBuffer.wrap(sessionBytes));
TraceState state = Tracing.instance.get(sessionId);
String message = String.format("Sending %s message to %s", messageOut.verb, toFull.address);
// session may have already finished; see CASSANDRA-5668
if (state == null)
{
byte[] traceTypeBytes = (byte[]) messageOut.parameters.get(Tracing.TRACE_TYPE);
Tracing.TraceType traceType = traceTypeBytes == null ? Tracing.TraceType.QUERY : Tracing.TraceType.deserialize(traceTypeBytes[0]);
Tracing.instance.trace(ByteBuffer.wrap(sessionBytes), message, traceType.getTTL());
}
else
{
state.trace(message);
if (messageOut.verb == MessagingService.Verb.REQUEST_RESPONSE)
Tracing.instance.doneWithNonLocalSession(state);
}
}
out.writeInt(MessagingService.PROTOCOL_MAGIC);
out.writeInt(id);
long timestamp = System.currentTimeMillis();
@ -319,6 +344,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
{
mkdirs();
assert config.networkTopology().contains(config.broadcastAddressAndPort());
DistributedTestSnitch.assign(config.networkTopology());
DatabaseDescriptor.daemonInitialization();
DatabaseDescriptor.createAllDirectories();
@ -371,6 +399,8 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
initializeRing(cluster);
}
StorageService.instance.ensureTraceKeyspace();
SystemKeyspace.finishStartup();
if (!FBUtilities.getBroadcastAddress().equals(broadcastAddressAndPort().address))

3
test/distributed/org/apache/cassandra/distributed/impl/InstanceClassLoader.java
View File

@ -37,7 +37,8 @@ public class InstanceClassLoader extends URLClassLoader
Pair.class,
InetAddressAndPort.class,
ParameterizedClass.class,
IInvokableInstance.class
IInvokableInstance.class,
NetworkTopology.class
})
.map(Class::getName)
.collect(Collectors.toSet());

32
test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java
View File

@ -24,7 +24,6 @@ import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInstanceConfig;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.SimpleSeedProvider;
import org.apache.cassandra.locator.SimpleSnitch;
import java.io.File;
import java.lang.reflect.Field;
@ -43,11 +42,14 @@ public class InstanceConfig implements IInstanceConfig
public final int num;
public int num() { return num; }
private final NetworkTopology networkTopology;
public NetworkTopology networkTopology() { return networkTopology; }
public final UUID hostId;
public UUID hostId() { return hostId; }
private final Map<String, Object> params = new TreeMap<>();
private EnumSet featureFlags;
private final EnumSet featureFlags;
private volatile InetAddressAndPort broadcastAddressAndPort;
@ -69,6 +71,7 @@ public class InstanceConfig implements IInstanceConfig
}
private InstanceConfig(int num,
NetworkTopology networkTopology,
String broadcast_address,
String listen_address,
String broadcast_rpc_address,
@ -81,6 +84,7 @@ public class InstanceConfig implements IInstanceConfig
String initial_token)
{
this.num = num;
this.networkTopology = networkTopology;
this.hostId = java.util.UUID.randomUUID();
this .set("broadcast_address", broadcast_address)
.set("listen_address", listen_address)
@ -101,8 +105,8 @@ public class InstanceConfig implements IInstanceConfig
.set("concurrent_compactors", 1)
.set("memtable_heap_space_in_mb", 10)
.set("commitlog_sync", "batch")
.set("storage_port", 7010)
.set("endpoint_snitch", SimpleSnitch.class.getName())
.set("storage_port", 7012)
.set("endpoint_snitch", DistributedTestSnitch.class.getName())
.set("seed_provider", new ParameterizedClass(SimpleSeedProvider.class.getName(),
Collections.singletonMap("seeds", "127.0.0.1")))
// legacy parameters
@ -113,9 +117,11 @@ public class InstanceConfig implements IInstanceConfig
private InstanceConfig(InstanceConfig copy)
{
this.num = copy.num;
this.networkTopology = new NetworkTopology(copy.networkTopology);
this.params.putAll(copy.params);
this.hostId = copy.hostId;
this.featureFlags = copy.featureFlags;
this.broadcastAddressAndPort = copy.broadcastAddressAndPort;
}
public InstanceConfig with(Feature featureFlag)
@ -192,11 +198,7 @@ public class InstanceConfig implements IInstanceConfig
{
valueField.set(writeToConfig, value);
}
catch (IllegalAccessException e)
{
throw new IllegalStateException(e);
}
catch (IllegalArgumentException e)
catch (IllegalAccessException | IllegalArgumentException e)
{
throw new IllegalStateException(e);
}
@ -217,14 +219,14 @@ public class InstanceConfig implements IInstanceConfig
return (String)params.get(name);
}
public static InstanceConfig generate(int nodeNum, int subnet, File root, String token)
public static InstanceConfig generate(int nodeNum, String ipAddress, NetworkTopology networkTopology, File root, String token)
{
String ipPrefix = "127.0." + subnet + ".";
return new InstanceConfig(nodeNum,
ipPrefix + nodeNum,
ipPrefix + nodeNum,
ipPrefix + nodeNum,
ipPrefix + nodeNum,
networkTopology,
ipAddress,
ipAddress,
ipAddress,
ipAddress,
String.format("%s/node%d/saved_caches", root, nodeNum),
new String[] { String.format("%s/node%d/data", root, nodeNum) },
String.format("%s/node%d/commitlog", root, nodeNum),

6
test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java
View File

@ -77,7 +77,7 @@ public class IsolatedExecutor implements IIsolatedExecutor
return t;
};
ExecutorService shutdownExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 0, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(), threadFactory);
new LinkedBlockingQueue<>(), threadFactory);
return shutdownExecutor.submit(() -> {
try
{
@ -165,7 +165,7 @@ public class IsolatedExecutor implements IIsolatedExecutor
public static Object deserializeOneObject(byte[] bytes)
{
try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
ObjectInputStream ois = new ObjectInputStream(bais);)
ObjectInputStream ois = new ObjectInputStream(bais))
{
return ois.readObject();
}
@ -190,7 +190,7 @@ public class IsolatedExecutor implements IIsolatedExecutor
}
}
private static <T> T waitOn(Future<T> f)
public static <T> T waitOn(Future<T> f)
{
try
{

89
test/distributed/org/apache/cassandra/distributed/impl/NetworkTopology.java Normal file
View File

@ -0,0 +1,89 @@
/*
* 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.impl;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.utils.Pair;
public class NetworkTopology
{
private final Map<InetAddressAndPort, Pair<String, String>> map;
private NetworkTopology() {
map = new HashMap<>();
}
@SuppressWarnings("WeakerAccess")
public NetworkTopology(NetworkTopology networkTopology)
{
map = new HashMap<>(networkTopology.map);
}
public static NetworkTopology build(String ipPrefix, int broadcastPort, Map<Integer, Pair<String, String>> nodeIdTopology)
{
final NetworkTopology topology = new NetworkTopology();
for (int nodeId = 1; nodeId <= nodeIdTopology.size(); nodeId++)
{
String broadcastAddress = ipPrefix + nodeId;
try
{
Pair<String,String> dcAndRack = nodeIdTopology.get(nodeId);
if (dcAndRack == null)
throw new IllegalStateException("nodeId " + nodeId + "not found in instanceMap");
InetAddressAndPort broadcastAddressAndPort = InetAddressAndPort.getByAddressOverrideDefaults(
InetAddress.getByName(broadcastAddress), broadcastPort);
topology.put(broadcastAddressAndPort, dcAndRack);
}
catch (UnknownHostException e)
{
throw new ConfigurationException("Unknown broadcast_address '" + broadcastAddress + '\'', false);
}
}
return topology;
}
public Pair<String, String> put(InetAddressAndPort key, Pair<String, String> value)
{
return map.put(key, value);
}
public String localRack(InetAddressAndPort key)
{
return map.get(key).right;
}
public String localDC(InetAddressAndPort key)
{
return map.get(key).left;
}
public boolean contains(InetAddressAndPort key)
{
return map.containsKey(key);
}
}

115
test/distributed/org/apache/cassandra/distributed/impl/TracingUtil.java Normal file
View File

@ -0,0 +1,115 @@
/*
* 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.impl;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.UUID;
import org.apache.cassandra.db.ConsistencyLevel;
/**
* Utilities for accessing the system_traces table from in-JVM dtests
*/
public class TracingUtil
{
/**
* Represents an entry from system_traces
*/
public static class TraceEntry
{
public final UUID sessionId;
public final UUID eventId;
public final String activity;
public final InetAddress source;
public final int sourceElapsed;
public final String thread;
private TraceEntry(UUID sessionId, UUID eventId, String activity, InetAddress sourceIP, int sourceElapsed, String thread)
{
this.sessionId = sessionId;
this.eventId = eventId;
this.activity = activity;
this.source = sourceIP;
this.sourceElapsed = sourceElapsed;
this.thread = thread;
}
static TraceEntry fromRowResultObjects(Object[] objects)
{
return new TraceEntry((UUID) objects[0],
(UUID) objects[1],
(String) objects[2],
(InetAddress) objects[3],
(Integer) objects[4],
(String) objects[5]);
}
}
public static List<TraceEntry> getTrace(AbstractCluster cluster, UUID sessionId)
{
return getTrace(cluster, sessionId, ConsistencyLevel.ALL);
}
public static List<TraceEntry> getTrace(AbstractCluster cluster, UUID sessionId, ConsistencyLevel cl)
{
Object[][] result = cluster.coordinator(1).execute(
"SELECT session_id, event_id, activity, source, source_elapsed, thread " +
"FROM system_traces.events WHERE session_id = ?", cl, sessionId);
List<TraceEntry> traces = new LinkedList<>();
for (Object[] r : result)
{
traces.add(TraceEntry.fromRowResultObjects(r));
}
return traces;
}
public static List<TraceEntry> getTraces(AbstractCluster cluster)
{
return getTraces(cluster, ConsistencyLevel.ALL);
}
public static List<TraceEntry> getTraces(AbstractCluster cluster, ConsistencyLevel cl)
{
Object[][] result = cluster.coordinator(1).execute(
"SELECT session_id, event_id, activity, source, source_elapsed, thread " +
"FROM system_traces.events", cl);
List<TraceEntry> traces = new ArrayList<>();
for (Object[] r : result)
{
traces.add(TraceEntry.fromRowResultObjects(r));
}
return traces;
}
// Set up the wait for tracing time system property, returning the previous value.
// Handles being called again to reset with the original value, replacing the null
// with the default value.
public static String setWaitForTracingEventTimeoutSecs(String timeoutInSeconds)
{
return System.setProperty("cassandra.wait_for_tracing_events_timeout_secs",
timeoutInSeconds == null ? "0" : timeoutInSeconds);
}
}

2
test/distributed/org/apache/cassandra/distributed/test/DistributedReadWritePathTest.java
View File

@ -85,7 +85,7 @@ public class DistributedReadWritePathTest extends DistributedTestBase
assertRows(cluster.get(3).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1"));
assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1",
ConsistencyLevel.QUORUM),
ConsistencyLevel.ALL), // ensure node3 in preflist
row(1, 1, 1));
// Verify that data got repaired to the third node

92
test/distributed/org/apache/cassandra/distributed/test/MessageForwardingTest.java Normal file
View File

@ -0,0 +1,92 @@
/*
* 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.io.IOException;
import java.net.InetAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Future;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.impl.IsolatedExecutor;
import org.apache.cassandra.distributed.impl.TracingUtil;
import org.apache.cassandra.utils.UUIDGen;
public class MessageForwardingTest extends DistributedTestBase
{
@Test
public void mutationsForwardedToAllReplicasTest()
{
String originalTraceTimeout = TracingUtil.setWaitForTracingEventTimeoutSecs("1");
final int numInserts = 100;
Map<InetAddress,Integer> commitCounts = new HashMap<>();
try (Cluster cluster = init(Cluster.build()
.withDC("dc0", 1)
.withDC("dc1", 3)
.start()))
{
cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v text, PRIMARY KEY (pk, ck))");
cluster.forEach(instance -> commitCounts.put(instance.broadcastAddressAndPort().address, 0));
final UUID sessionId = UUIDGen.getTimeUUID();
Stream<Future<Object[][]>> inserts = IntStream.range(0, numInserts).mapToObj((idx) ->
cluster.coordinator(1).asyncExecuteWithTracing(sessionId,
"INSERT INTO " + KEYSPACE + ".tbl(pk,ck,v) VALUES (1, 1, 'x')",
ConsistencyLevel.ALL)
);
// Wait for each of the futures to complete before checking the traces, don't care
// about the result so
//noinspection ResultOfMethodCallIgnored
inserts.map(IsolatedExecutor::waitOn).count();
cluster.forEach(instance -> commitCounts.put(instance.broadcastAddressAndPort().address, 0));
List<TracingUtil.TraceEntry> traces = TracingUtil.getTrace(cluster, sessionId, ConsistencyLevel.ALL);
traces.forEach(traceEntry -> {
if (traceEntry.activity.contains("Appending to commitlog"))
{
commitCounts.compute(traceEntry.source, (k, v) -> (v != null ? v : 0) + 1);
}
});
// Check that each node received the forwarded messages once (and only once)
commitCounts.forEach((source, count) ->
Assert.assertEquals(source + " appending to commitlog traces",
(long) numInserts, (long) count));
}
catch (IOException e)
{
Assert.fail("Threw exception: " + e);
}
finally
{
TracingUtil.setWaitForTracingEventTimeoutSecs(originalTraceTimeout);
}
}
}

99
test/distributed/org/apache/cassandra/distributed/test/NetworkTopologyTest.java Normal file
View File

@ -0,0 +1,99 @@
/*
* 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.Collections;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.IInstance;
import org.apache.cassandra.utils.Pair;
public class NetworkTopologyTest extends DistributedTestBase
{
@Test
public void namedDcTest() throws Throwable
{
try (Cluster cluster = Cluster.build()
.withNodeIdTopology(Collections.singletonMap(1, Pair.create("somewhere", "rack0")))
.withRack("elsewhere", "firstrack", 1)
.withRack("elsewhere", "secondrack", 2)
.withDC("nearthere", 4)
.start())
{
Assert.assertEquals(1, cluster.stream("somewhere").count());
Assert.assertEquals(1, cluster.stream("elsewhere", "firstrack").count());
Assert.assertEquals(2, cluster.stream("elsewhere", "secondrack").count());
Assert.assertEquals(3, cluster.stream("elsewhere").count());
Assert.assertEquals(4, cluster.stream("nearthere").count());
Set<IInstance> expect = cluster.stream().collect(Collectors.toSet());
Set<IInstance> result = Stream.concat(Stream.concat(cluster.stream("somewhere"),
cluster.stream("elsewhere")),
cluster.stream("nearthere")).collect(Collectors.toSet());
Assert.assertEquals(expect, result);
}
}
@Test
public void automaticNamedDcTest() throws Throwable
{
try (Cluster cluster = Cluster.build()
.withRacks(2, 1, 3)
.start())
{
Assert.assertEquals(6, cluster.stream().count());
Assert.assertEquals(3, cluster.stream("datacenter1").count());
Assert.assertEquals(3, cluster.stream("datacenter2", "rack1").count());
}
}
@Test(expected = IllegalStateException.class)
public void noCountsAfterNamingDCsTest()
{
Cluster.build()
.withDC("nameddc", 1)
.withDCs(1);
}
@Test(expected = IllegalStateException.class)
public void mustProvideNodeCountBeforeWithDCsTest()
{
Cluster.build()
.withDCs(1);
}
@Test(expected = IllegalStateException.class)
public void noEmptyNodeIdTopologyTest()
{
Cluster.build().withNodeIdTopology(Collections.emptyMap());
}
@Test(expected = IllegalStateException.class)
public void noHolesInNodeIdTopologyTest()
{
Cluster.build().withNodeIdTopology(Collections.singletonMap(2, Pair.create("doomed", "rack")));
}
}