You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by if...@apache.org on 2020/02/05 10:57:21 UTC
[cassandra] branch trunk updated (342d160 -> a041b53)
This is an automated email from the ASF dual-hosted git repository.
ifesdjeen pushed a change to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git.
from 342d160 Merge branch 'cassandra-3.11' into trunk
new 9705d82 Add an ability to run bootstrap / streaming tests with in-JVM dtest framework.
new 3320f08 Merge branch 'cassandra-2.2' into cassandra-3.0
new c09e709 Merge branch 'cassandra-3.0' into cassandra-3.11
new a041b53 Merge branch 'cassandra-3.11' into trunk
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/cassandra/service/GCInspector.java | 6 +-
.../cassandra/distributed/api/IInstanceConfig.java | 4 +
.../distributed/impl/AbstractCluster.java | 172 +++++++++++++++------
.../distributed/impl/DistributedTestSnitch.java | 52 ++++++-
.../cassandra/distributed/impl/InstanceConfig.java | 10 +-
.../distributed/impl/NetworkTopology.java | 60 ++++++-
.../cassandra/distributed/test/BootstrapTest.java | 104 +++++++++++++
.../distributed/test/DistributedTestBase.java | 1 +
.../distributed/test/NetworkTopologyTest.java | 5 +-
.../cassandra/service/LeaveAndBootstrapTest.java | 27 ++--
10 files changed, 363 insertions(+), 78 deletions(-)
create mode 100644 test/distributed/org/apache/cassandra/distributed/test/BootstrapTest.java
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
[cassandra] 01/01: Merge branch 'cassandra-3.11' into trunk
Posted by if...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ifesdjeen pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit a041b53d2996cb076f615bbe3dfc048ed77372d4
Merge: 342d160 c09e709
Author: Alex Petrov <ol...@gmail.com>
AuthorDate: Wed Feb 5 11:55:18 2020 +0100
Merge branch 'cassandra-3.11' into trunk
.../org/apache/cassandra/service/GCInspector.java | 6 +-
.../cassandra/distributed/api/IInstanceConfig.java | 4 +
.../distributed/impl/AbstractCluster.java | 172 +++++++++++++++------
.../distributed/impl/DistributedTestSnitch.java | 52 ++++++-
.../cassandra/distributed/impl/InstanceConfig.java | 10 +-
.../distributed/impl/NetworkTopology.java | 60 ++++++-
.../cassandra/distributed/test/BootstrapTest.java | 104 +++++++++++++
.../distributed/test/DistributedTestBase.java | 1 +
.../distributed/test/NetworkTopologyTest.java | 5 +-
.../cassandra/service/LeaveAndBootstrapTest.java | 27 ++--
10 files changed, 363 insertions(+), 78 deletions(-)
diff --cc src/java/org/apache/cassandra/service/GCInspector.java
index e0a935d,787d79a..02fd720
--- a/src/java/org/apache/cassandra/service/GCInspector.java
+++ b/src/java/org/apache/cassandra/service/GCInspector.java
@@@ -160,12 -147,10 +161,11 @@@ public class GCInspector implements Not
GarbageCollectorMXBean gc = ManagementFactory.newPlatformMXBeanProxy(mbs, name.getCanonicalName(), GarbageCollectorMXBean.class);
gcStates.put(gc.getName(), new GCState(gc, assumeGCIsPartiallyConcurrent(gc), assumeGCIsOldGen(gc)));
}
-
- MBeanWrapper.instance.registerMBean(this, new ObjectName(MBEAN_NAME));
+ ObjectName me = new ObjectName(MBEAN_NAME);
+ if (!mbs.isRegistered(me))
- mbs.registerMBean(this, new ObjectName(MBEAN_NAME));
++ MBeanWrapper.instance.registerMBean(this, new ObjectName(MBEAN_NAME));
}
- catch (RuntimeException | InstanceAlreadyExistsException | MBeanRegistrationException |
- NotCompliantMBeanException | MalformedObjectNameException | IOException e)
- catch (Exception e)
++ catch (MalformedObjectNameException | IOException e)
{
throw new RuntimeException(e);
}
diff --cc test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
index d487c14,400637d..f03bae0
--- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
@@@ -58,9 -57,8 +58,8 @@@ import org.apache.cassandra.distributed
import org.apache.cassandra.distributed.api.IMessageFilters;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.Verb;
import org.apache.cassandra.utils.FBUtilities;
- import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
/**
@@@ -102,9 -100,10 +101,11 @@@ public abstract class AbstractCluster<
private final List<I> instances;
private final Map<InetAddressAndPort, I> instanceMap;
+ private final Versions.Version initialVersion;
+
// mutated by user-facing API
private final MessageFilters filters;
+ private volatile Thread.UncaughtExceptionHandler previousHandler = null;
protected class Wrapper extends DelegatingInvokableInstance implements IUpgradeableInstance
{
@@@ -133,8 -132,8 +134,8 @@@
private IInvokableInstance newInstance(int generation)
{
ClassLoader classLoader = new InstanceClassLoader(generation, config.num, version.classpath, sharedClassLoader);
- return Instance.transferAdhoc((SerializableBiFunction<IInstanceConfig, ClassLoader, Instance>)Instance::new, classLoader)
- .apply(config.forVersion(version.major), classLoader);
+ return Instance.transferAdhoc((SerializableBiFunction<IInstanceConfig, ClassLoader, Instance>) Instance::new, classLoader)
- .apply(config, classLoader);
++ .apply(config.forVersion(version.major), classLoader);
}
public IInstanceConfig config()
@@@ -218,18 -208,9 +219,18 @@@
delegate = null;
}
}
+
+ public void uncaughtException(Thread thread, Throwable throwable)
+ {
+ IInvokableInstance delegate = this.delegate;
+ if (delegate != null)
+ delegate.uncaughtException(thread, throwable);
+ else
+ logger.error("uncaught exception in thread {}", thread, throwable);
+ }
}
- protected AbstractCluster(File root, Versions.Version version, List<InstanceConfig> configs,
+ protected AbstractCluster(File root, Versions.Version initialVersion, List<InstanceConfig> configs,
ClassLoader sharedClassLoader)
{
this.root = root;
diff --cc test/distributed/org/apache/cassandra/distributed/impl/DistributedTestSnitch.java
index 35e2903,9147169..736e77b
--- a/test/distributed/org/apache/cassandra/distributed/impl/DistributedTestSnitch.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/DistributedTestSnitch.java
@@@ -19,10 -19,18 +19,17 @@@
package org.apache.cassandra.distributed.impl;
import java.net.InetAddress;
-import java.util.HashMap;
+ import java.util.Map;
import org.apache.cassandra.config.Config;
+ import org.apache.cassandra.db.SystemKeyspace;
+ import org.apache.cassandra.gms.ApplicationState;
+ import org.apache.cassandra.gms.EndpointState;
+ import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.locator.AbstractNetworkTopologySnitch;
import org.apache.cassandra.locator.InetAddressAndPort;
+ import org.apache.cassandra.service.StorageService;
+ import org.apache.cassandra.utils.FBUtilities;
public class DistributedTestSnitch extends AbstractNetworkTopologySnitch
{
@@@ -51,11 -61,49 +60,42 @@@
public String getDatacenter(InetAddressAndPort endpoint)
{
assert mapping != null : "network topology must be assigned before using snitch";
- return mapping.localDC(endpoint);
+ return maybeGetFromEndpointState(mapping.localDC(endpoint), endpoint, ApplicationState.DC, DEFAULT_DC);
+ }
+
+ // Here, the logic is slightly different from what we have in GossipingPropertyFileSnitch since we have a different
+ // goal. Passed argument (topology that was set on the node) overrides anything that is passed elsewhere.
+ private String maybeGetFromEndpointState(String current, InetAddressAndPort endpoint, ApplicationState state, String defaultValue)
+ {
+ if (current != null)
+ return current;
+
- EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(endpoint.address);
++ EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
+ if (epState == null || epState.getApplicationState(state) == null)
+ {
+ if (savedEndpoints == null)
- {
- savedEndpoints = new HashMap<>();
- int storage_port = Config.getOverrideLoadConfig().get().storage_port;
- for (Map.Entry<InetAddress, Map<String, String>> entry : SystemKeyspace.loadDcRackInfo().entrySet())
- {
- savedEndpoints.put(InetAddressAndPort.getByAddressOverrideDefaults(endpoint.address, storage_port),
- entry.getValue());
- }
- }
++ savedEndpoints = SystemKeyspace.loadDcRackInfo();
+ if (savedEndpoints.containsKey(endpoint))
+ return savedEndpoints.get(endpoint).get("data_center");
+
+ return defaultValue;
+ }
+
+ return epState.getApplicationState(state).value;
}
static void assign(NetworkTopology newMapping)
{
mapping = new NetworkTopology(newMapping);
}
+
+ public void gossiperStarting()
+ {
+ super.gossiperStarting();
+
-
++ Gossiper.instance.addLocalApplicationState(ApplicationState.INTERNAL_ADDRESS_AND_PORT,
++ StorageService.instance.valueFactory.internalAddressAndPort(FBUtilities.getLocalAddressAndPort()));
+ Gossiper.instance.addLocalApplicationState(ApplicationState.INTERNAL_IP,
- StorageService.instance.valueFactory.internalIP(FBUtilities.getLocalAddress().getHostAddress()));
++ StorageService.instance.valueFactory.internalIP(FBUtilities.getJustLocalAddress().getHostAddress()));
+ }
}
diff --cc test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java
index 3c4748e,c5f198e..018e3ea
--- a/test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java
+++ b/test/unit/org/apache/cassandra/service/LeaveAndBootstrapTest.java
@@@ -90,35 -94,32 +90,32 @@@ public class LeaveAndBootstrapTes
IPartitioner partitioner = RandomPartitioner.instance;
VersionedValue.VersionedValueFactory valueFactory = new VersionedValue.VersionedValueFactory(partitioner);
-- ArrayList<Token> endpointTokens = new ArrayList<Token>();
-- ArrayList<Token> keyTokens = new ArrayList<Token>();
- List<InetAddress> hosts = new ArrayList<InetAddress>();
- List<UUID> hostIds = new ArrayList<UUID>();
++ ArrayList<Token> endpointTokens = new ArrayList<>();
++ ArrayList<Token> keyTokens = new ArrayList<>();
+ List<InetAddressAndPort> hosts = new ArrayList<>();
- List<UUID> hostIds = new ArrayList<UUID>();
++ List<UUID> hostIds = new ArrayList<>();
Util.createInitialRing(ss, partitioner, endpointTokens, keyTokens, hosts, hostIds, RING_SIZE);
- Map<Token, List<InetAddressAndPort>> expectedEndpoints = new HashMap<Token, List<InetAddressAndPort>>();
- Map<Token, List<InetAddress>> expectedEndpoints = new HashMap<Token, List<InetAddress>>();
-- for (String keyspaceName : Schema.instance.getNonLocalStrategyKeyspaces())
++ Map<Token, List<InetAddressAndPort>> expectedEndpoints = new HashMap<>();
++ for (Token token : keyTokens)
{
-- for (Token token : keyTokens)
++ List<InetAddressAndPort> endpoints = new ArrayList<>();
++ Iterator<Token> tokenIter = TokenMetadata.ringIterator(tmd.sortedTokens(), token, false);
++ while (tokenIter.hasNext())
{
- List<InetAddressAndPort> endpoints = new ArrayList<>();
- List<InetAddress> endpoints = new ArrayList<InetAddress>();
-- Iterator<Token> tokenIter = TokenMetadata.ringIterator(tmd.sortedTokens(), token, false);
-- while (tokenIter.hasNext())
-- {
-- endpoints.add(tmd.getEndpoint(tokenIter.next()));
-- }
-- expectedEndpoints.put(token, endpoints);
++ endpoints.add(tmd.getEndpoint(tokenIter.next()));
}
++ expectedEndpoints.put(token, endpoints);
}
// Third node leaves
ss.onChange(hosts.get(LEAVING_NODE),
- ApplicationState.STATUS,
- valueFactory.leaving(Collections.singleton(endpointTokens.get(LEAVING_NODE))));
+ ApplicationState.STATUS_WITH_PORT,
+ valueFactory.leaving(Collections.singleton(endpointTokens.get(LEAVING_NODE))));
+ ss.onChange(hosts.get(LEAVING_NODE),
- ApplicationState.STATUS,
- valueFactory.leaving(Collections.singleton(endpointTokens.get(LEAVING_NODE))));
++ ApplicationState.STATUS,
++ valueFactory.leaving(Collections.singleton(endpointTokens.get(LEAVING_NODE))));
assertTrue(tmd.isLeaving(hosts.get(LEAVING_NODE)));
Thread.sleep(100); // because there is a tight race between submit and blockUntilFinished
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org