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

Merge branch 'cassandra-2.2' into cassandra-3.0

This commit is contained in:
Benedict Elliott Smith 2021-11-10 11:10:04 +00:00
parent 969531a113 4884e9b58f
commit e4d96721bf
5 changed files with 93 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
build.xml
View File

@ -348,7 +348,7 @@
<exclusion groupId="commons-lang" artifactId="commons-lang"/>
</dependency>
<dependency groupId="org.mockito" artifactId="mockito-core" version="3.2.4" scope="test"/>
<dependency groupId="org.apache.cassandra" artifactId="dtest-api" version="0.0.9" scope="test"/>
<dependency groupId="org.apache.cassandra" artifactId="dtest-api" version="0.0.11" scope="test"/>
<dependency groupId="org.reflections" artifactId="reflections" version="0.9.12" scope="test"/>
<dependency groupId="org.quicktheories" artifactId="quicktheories" version="0.25" scope="test"/>
<dependency groupId="org.apache.hadoop" artifactId="hadoop-core" version="1.0.3" scope="provided">

50
test/distributed/org/apache/cassandra/distributed/impl/DelegatingInvokableInstance.java
View File

@ -20,7 +20,6 @@ package org.apache.cassandra.distributed.impl;
import java.io.Serializable;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Future;
import java.util.function.BiConsumer;
@ -35,8 +34,6 @@ import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.IListen;
import org.apache.cassandra.distributed.api.IMessage;
import org.apache.cassandra.distributed.api.SimpleQueryResult;
import org.apache.cassandra.distributed.shared.NetworkTopology;
import org.apache.cassandra.utils.FBUtilities;
public abstract class DelegatingInvokableInstance implements IInvokableInstance
{
@ -142,12 +139,24 @@ public abstract class DelegatingInvokableInstance implements IInvokableInstance
delegateForStartup().startup(cluster);
}
@Override
public void postStartup()
{
delegateForStartup().postStartup();
}
@Override
public void receiveMessage(IMessage message)
{
delegate().receiveMessage(message);
}
@Override
public void receiveMessageWithInvokingThread(IMessage message)
{
delegate().receiveMessageWithInvokingThread(message);
}
@Override
public <O> CallableNoExcept<Future<O>> async(CallableNoExcept<O> call)
{
@ -196,6 +205,18 @@ public abstract class DelegatingInvokableInstance implements IInvokableInstance
return delegate().sync(consumer);
}
@Override
public <I1, I2, I3> TriFunction<I1, I2, I3, Future<?>> async(TriConsumer<I1, I2, I3> consumer)
{
return delegate().async(consumer);
}
@Override
public <I1, I2, I3> TriConsumer<I1, I2, I3> sync(TriConsumer<I1, I2, I3> consumer)
{
return delegate().sync(consumer);
}
@Override
public <I, O> Function<I, Future<O>> async(Function<I, O> f)
{
@ -232,4 +253,27 @@ public abstract class DelegatingInvokableInstance implements IInvokableInstance
return delegate().sync(f);
}
@Override
public <I1, I2, I3, I4, O> QuadFunction<I1, I2, I3, I4, Future<O>> async(QuadFunction<I1, I2, I3, I4, O> f)
{
return delegate().async(f);
}
@Override
public <I1, I2, I3, I4, O> QuadFunction<I1, I2, I3, I4, O> sync(QuadFunction<I1, I2, I3, I4, O> f)
{
return delegate().sync(f);
}
@Override
public <I1, I2, I3, I4, I5, O> QuintFunction<I1, I2, I3, I4, I5, Future<O>> async(QuintFunction<I1, I2, I3, I4, I5, O> f)
{
return delegate().async(f);
}
@Override
public <I1, I2, I3, I4, I5, O> QuintFunction<I1, I2, I3, I4, I5, O> sync(QuintFunction<I1, I2, I3, I4, I5, O> f)
{
return delegate().sync(f);
}
}

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

@ -470,40 +470,44 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
public void receiveMessage(IMessage imessage)
{
sync(() -> {
Pair<MessageIn<Object>, Integer> deserialized = null;
try
{
deserialized = deserializeMessage(imessage);
}
catch (Throwable t)
{
throw new RuntimeException("Exception occurred on node " + broadcastAddress(), t);
}
sync(() -> receiveMessageWithInvokingThread(imessage)).run();
}
MessageIn<Object> message = deserialized.left;
int partial = deserialized.right;
@Override
public void receiveMessageWithInvokingThread(IMessage imessage)
{
Pair<MessageIn<Object>, Integer> deserialized = null;
try
{
deserialized = deserializeMessage(imessage);
}
catch (Throwable t)
{
throw new RuntimeException("Exception occurred on node " + broadcastAddress(), t);
}
long timestamp = System.currentTimeMillis();
boolean isCrossNodeTimestamp = false;
MessageIn<Object> message = deserialized.left;
int partial = deserialized.right;
if (DatabaseDescriptor.hasCrossNodeTimeout())
{
long crossNodeTimestamp = (timestamp & 0xFFFFFFFF00000000L) | (((partial & 0xFFFFFFFFL) << 2) >> 2);
isCrossNodeTimestamp = (timestamp != crossNodeTimestamp);
timestamp = crossNodeTimestamp;
}
long timestamp = System.currentTimeMillis();
boolean isCrossNodeTimestamp = false;
if (message == null)
{
// callback expired; nothing to do
return;
}
if (message.version <= MessagingService.current_version)
{
MessagingService.instance().receive(message, imessage.id(), timestamp, isCrossNodeTimestamp);
}
}).run();
if (DatabaseDescriptor.hasCrossNodeTimeout())
{
long crossNodeTimestamp = (timestamp & 0xFFFFFFFF00000000L) | (((partial & 0xFFFFFFFFL) << 2) >> 2);
isCrossNodeTimestamp = (timestamp != crossNodeTimestamp);
timestamp = crossNodeTimestamp;
}
if (message == null)
{
// callback expired; nothing to do
return;
}
if (message.version <= MessagingService.current_version)
{
MessagingService.instance().receive(message, imessage.id(), timestamp, isCrossNodeTimestamp);
}
}
public int getMessagingVersion()
@ -648,7 +652,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
private static Config loadConfig(IInstanceConfig overrides)
{
Map<String,Object> params = ((InstanceConfig) overrides).getParams();
Map<String,Object> params = overrides.getParams();
boolean check = true;
if (overrides.get(Constants.KEY_DTEST_API_CONFIG_CHECK) != null)
check = (boolean) overrides.get(Constants.KEY_DTEST_API_CONFIG_CHECK);

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

@ -36,7 +36,6 @@ import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.IInstanceConfig;
import org.apache.cassandra.distributed.shared.NetworkTopology;
import org.apache.cassandra.distributed.shared.Shared;
import org.apache.cassandra.distributed.upgrade.UpgradeTestBase;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.SimpleSeedProvider;
@ -189,7 +188,7 @@ public class InstanceConfig implements IInstanceConfig
return this;
}
private InstanceConfig forceSet(String fieldName, Object value)
public InstanceConfig forceSet(String fieldName, Object value)
{
if (value == null)
value = NULL;

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

@ -117,6 +117,9 @@ public class IsolatedExecutor implements IIsolatedExecutor
public <I1, I2> BiFunction<I1, I2, Future<?>> async(BiConsumer<I1, I2> consumer) { return (a, b) -> isolatedExecutor.submit(() -> consumer.accept(a, b)); }
public <I1, I2> BiConsumer<I1, I2> sync(BiConsumer<I1, I2> consumer) { return (a, b) -> waitOn(async(consumer).apply(a, b)); }
public <I1, I2, I3> TriFunction<I1, I2, I3, Future<?>> async(TriConsumer<I1, I2, I3> consumer) { return (a, b, c) -> isolatedExecutor.submit(() -> consumer.accept(a, b, c)); }
public <I1, I2, I3> TriConsumer<I1, I2, I3> sync(TriConsumer<I1, I2, I3> consumer) { return (a, b, c) -> waitOn(async(consumer).apply(a, b, c)); }
public <I, O> Function<I, Future<O>> async(Function<I, O> f) { return (a) -> isolatedExecutor.submit(() -> f.apply(a)); }
public <I, O> Function<I, O> sync(Function<I, O> f) { return (a) -> waitOn(async(f).apply(a)); }
@ -126,6 +129,12 @@ public class IsolatedExecutor implements IIsolatedExecutor
public <I1, I2, I3, O> TriFunction<I1, I2, I3, Future<O>> async(TriFunction<I1, I2, I3, O> f) { return (a, b, c) -> isolatedExecutor.submit(() -> f.apply(a, b, c)); }
public <I1, I2, I3, O> TriFunction<I1, I2, I3, O> sync(TriFunction<I1, I2, I3, O> f) { return (a, b, c) -> waitOn(async(f).apply(a, b, c)); }
public <I1, I2, I3, I4, O> QuadFunction<I1, I2, I3, I4, Future<O>> async(QuadFunction<I1, I2, I3, I4, O> f) { return (a, b, c, d) -> isolatedExecutor.submit(() -> f.apply(a, b, c, d)); }
public <I1, I2, I3, I4, O> QuadFunction<I1, I2, I3, I4, O> sync(QuadFunction<I1, I2, I3, I4, O> f) { return (a, b, c, d) -> waitOn(async(f).apply(a, b, c, d)); }
public <I1, I2, I3, I4, I5, O> QuintFunction<I1, I2, I3, I4, I5, Future<O>> async(QuintFunction<I1, I2, I3, I4, I5, O> f) { return (a, b, c, d, e) -> isolatedExecutor.submit(() -> f.apply(a, b, c, d, e)); }
public <I1, I2, I3, I4, I5, O> QuintFunction<I1, I2, I3, I4, I5, O> sync(QuintFunction<I1, I2, I3, I4, I5, O> f) { return (a, b, c, d,e ) -> waitOn(async(f).apply(a, b, c, d, e)); }
public <E extends Serializable> E transfer(E object)
{
return (E) transferOneObject(object, classLoader, deserializeOnInstance);