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:
Alex Petrov 2021-01-28 15:26:28 +01:00
parent 11b07536a9 88e7430bad
commit 24d133ecbd
2 changed files with 20 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -497,10 +497,12 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
instance.sync(() -> {
try (SchemaChangeMonitor monitor = new SchemaChangeMonitor())
{
// execute the schema change
instance.coordinator().execute(query, ConsistencyLevel.ALL);
if (ignoreStoppedInstances)
monitor.ignoreStoppedInstances();
monitor.startPolling();
// execute the schema change
instance.coordinator().execute(query, ConsistencyLevel.ALL);
monitor.waitForCompletion();
}
}).run();
@ -536,7 +538,7 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
private final long timeOut;
private final TimeUnit timeoutUnit;
protected Predicate<IInstance> instanceFilter;
volatile boolean changed;
volatile boolean initialized;
public ChangeMonitor(long timeOut, TimeUnit timeoutUnit)
{
@ -554,7 +556,7 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
protected void signal()
{
if (changed && isCompleted())
if (initialized && !completed.isSignaled() && isCompleted())
completed.signalAll();
}
@ -567,21 +569,20 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
public void waitForCompletion()
{
startPolling();
changed = true;
initialized = true;
signal();
try
{
if (!completed.await(timeOut, timeoutUnit))
throw new InterruptedException();
throw new IllegalStateException(getMonitorTimeoutMessage());
}
catch (InterruptedException e)
{
throw new IllegalStateException(getMonitorTimeoutMessage());
throw new IllegalStateException("Caught exception while waiting for completion", e);
}
}
private void startPolling()
protected void startPolling()
{
instances.stream().filter(instanceFilter).forEach(instance -> cleanup.add(startPolling(instance)));
}
@ -623,7 +624,8 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
protected String getMonitorTimeoutMessage()
{
return "Schema agreement not reached";
return String.format("Schema agreement not reached. Schema versions of the instances: %s",
instances.stream().map(IInstance::schemaVersion).collect(Collectors.toList()));
}
}
@ -656,6 +658,8 @@ public abstract class AbstractCluster<I extends IInstance> implements ICluster<I
Thread.setDefaultUncaughtExceptionHandler(this::uncaughtExceptions);
try (AllMembersAliveMonitor monitor = new AllMembersAliveMonitor())
{
monitor.startPolling();
// Start any instances with auto_bootstrap enabled first, and in series to avoid issues
// with multiple nodes bootstrapping with consistent range movement enabled,
// and then start any instances with it disabled in parallel.

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

@ -128,6 +128,7 @@ import static org.apache.cassandra.distributed.api.Feature.NETWORK;
public class Instance extends IsolatedExecutor implements IInvokableInstance
{
public final IInstanceConfig config;
private volatile boolean initialized = false;
// should never be invoked directly, so that it is instantiated on other class loader;
// only visible for inheritance
@ -584,6 +585,8 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
throw new RuntimeException(t);
}
}).run();
initialized = true;
}
private void mkdirs()
@ -747,6 +750,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
public int liveMemberCount()
{
if (!initialized || isShutdown())
return 0;
return sync(() -> {
if (!DatabaseDescriptor.isDaemonInitialized() || !Gossiper.instance.isEnabled())
return 0;