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:
Benedict Elliott Smith 2021-11-10 11:11:43 +00:00
parent 3d74cad35f e4d96721bf
commit bfe4a90852
5 changed files with 80 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
build.xml
View File

@ -370,7 +370,7 @@
<exclusion groupId="org.hamcrest" artifactId="hamcrest-core"/>
</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">

48
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;
@ -141,12 +140,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)
{
@ -195,6 +206,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)
{
@ -236,4 +259,27 @@ public abstract class DelegatingInvokableInstance implements IInvokableInstance
{
return delegate().callOnInstance(call);
}
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);
}
}

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

@ -438,27 +438,31 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
public void receiveMessage(IMessage imessage)
{
sync(() -> {
// Based on org.apache.cassandra.net.IncomingTcpConnection.receiveMessage
try
sync(() -> receiveMessageWithInvokingThread(imessage)).run();
}
@Override
public void receiveMessageWithInvokingThread(IMessage imessage)
{
// Based on org.apache.cassandra.net.IncomingTcpConnection.receiveMessage
try
{
MessageIn message = deserializeMessage(imessage);
if (message == null)
{
MessageIn message = deserializeMessage(imessage);
if (message == null)
{
// callback expired; nothing to do
return;
}
if (message.version <= MessagingService.current_version)
{
MessagingService.instance().receive(message, imessage.id());
}
// else ignore message
// callback expired; nothing to do
return;
}
catch (Throwable t)
if (message.version <= MessagingService.current_version)
{
throw new RuntimeException("Exception occurred on node " + broadcastAddress(), t);
MessagingService.instance().receive(message, imessage.id());
}
}).run();
// else ignore message
}
catch (Throwable t)
{
throw new RuntimeException("Exception occurred on node " + broadcastAddress(), t);
}
}
public int getMessagingVersion()
@ -603,7 +607,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);