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

Merge branch 'cassandra-4.0' into trunk

This commit is contained in:
Jon Meredith 2022-04-08 14:39:37 -06:00
parent 2fde9af74f ce6a65cb29
commit e3b3a59c6c
8 changed files with 63 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -133,6 +133,7 @@
* GossiperTest.testHasVersion3Nodes didn't take into account trunk version changes, fixed to rely on latest version (CASSANDRA-16651)
* Update JNA library to 5.9.0 and snappy-java to version 1.1.8.4 (CASSANDRA-17040)
Merged from 4.0:
* Clean up schema migration coordinator and tests (CASSANDRA-17533)
* Shut repair task executor down without interruption to avoid compromising shared channel proxies (CASSANDRA-17466)
* Generate valid KEYSPACE / MATERIALIZED VIEW for CQL for views (CASSANDRA-17266)
* Fix timestamp tz parsing (CASSANDRA-17467)

4
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
View File

@ -158,6 +158,10 @@ public enum CassandraRelevantProperties
GOSSIPER_SKIP_WAITING_TO_SETTLE("cassandra.skip_wait_for_gossip_to_settle", "-1"),
IGNORED_SCHEMA_CHECK_VERSIONS("cassandra.skip_schema_check_for_versions"),
IGNORED_SCHEMA_CHECK_ENDPOINTS("cassandra.skip_schema_check_for_endpoints"),
SHUTDOWN_ANNOUNCE_DELAY_IN_MS("cassandra.shutdown_announce_in_ms", "2000"),
/**

3
src/java/org/apache/cassandra/schema/DefaultSchemaUpdateHandler.java
View File

@ -116,7 +116,8 @@ public class DefaultSchemaUpdateHandler implements SchemaUpdateHandler, IEndpoin
Schema.instance.getVersion(),
migrationCoordinator.outstandingVersions(),
CassandraRelevantProperties.BOOTSTRAP_SKIP_SCHEMA_CHECK.getKey(),
MigrationCoordinator.IGNORED_ENDPOINTS_PROP, MigrationCoordinator.IGNORED_VERSIONS_PROP);
CassandraRelevantProperties.IGNORED_SCHEMA_CHECK_ENDPOINTS.getKey(),
CassandraRelevantProperties.IGNORED_SCHEMA_CHECK_VERSIONS.getKey());
if (requireSchemas)
{

44
src/java/org/apache/cassandra/schema/MigrationCoordinator.java
View File

@ -35,9 +35,11 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.LongSupplier;
import java.util.function.Supplier;
import com.google.common.annotations.VisibleForTesting;
@ -71,6 +73,8 @@ import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.ImmediateFuture;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static org.apache.cassandra.config.CassandraRelevantProperties.IGNORED_SCHEMA_CHECK_ENDPOINTS;
import static org.apache.cassandra.config.CassandraRelevantProperties.IGNORED_SCHEMA_CHECK_VERSIONS;
import static org.apache.cassandra.net.Verb.SCHEMA_PUSH_REQ;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.Simulate.With.MONITORS;
@ -90,15 +94,20 @@ public class MigrationCoordinator
private static final Logger logger = LoggerFactory.getLogger(MigrationCoordinator.class);
private static final Future<Void> FINISHED_FUTURE = ImmediateFuture.success(null);
private static LongSupplier getUptimeFn = () -> ManagementFactory.getRuntimeMXBean().getUptime();
@VisibleForTesting
public static void setUptimeFn(LongSupplier supplier)
{
getUptimeFn = supplier;
}
private static final int MIGRATION_DELAY_IN_MS = CassandraRelevantProperties.MIGRATION_DELAY.getInt();
public static final int MAX_OUTSTANDING_VERSION_REQUESTS = 3;
public static final String IGNORED_VERSIONS_PROP = "cassandra.skip_schema_check_for_versions";
public static final String IGNORED_ENDPOINTS_PROP = "cassandra.skip_schema_check_for_endpoints";
private static ImmutableSet<UUID> getIgnoredVersions()
{
String s = System.getProperty(IGNORED_VERSIONS_PROP);
String s = IGNORED_SCHEMA_CHECK_VERSIONS.getString();
if (s == null || s.isEmpty())
return ImmutableSet.of();
@ -117,7 +126,7 @@ public class MigrationCoordinator
{
Set<InetAddressAndPort> endpoints = new HashSet<>();
String s = System.getProperty(IGNORED_ENDPOINTS_PROP);
String s = IGNORED_SCHEMA_CHECK_ENDPOINTS.getString();
if (s == null || s.isEmpty())
return endpoints;
@ -335,7 +344,7 @@ public class MigrationCoordinator
private boolean shouldPullImmediately(InetAddressAndPort endpoint, UUID version)
{
UUID localSchemaVersion = schemaVersion.get();
if (SchemaConstants.emptyVersion.equals(localSchemaVersion) || ManagementFactory.getRuntimeMXBean().getUptime() < MIGRATION_DELAY_IN_MS)
if (SchemaConstants.emptyVersion.equals(localSchemaVersion) || getUptimeFn.getAsLong() < MIGRATION_DELAY_IN_MS)
{
// If we think we may be bootstrapping or have recently started, submit MigrationTask immediately
logger.debug("Immediately submitting migration task for {}, " +
@ -467,16 +476,29 @@ public class MigrationCoordinator
SchemaDiagnostics.versionAnnounced(Schema.instance);
}
private Future<Void> submitToMigrationIfNotShutdown(Runnable task)
private Future<?> submitToMigrationIfNotShutdown(Runnable task)
{
if (executor.isShutdown() || executor.isTerminated())
boolean skipped = false;
try
{
logger.info("Skipped scheduled pulling schema from other nodes: the MIGRATION executor service has been shutdown.");
if (executor.isShutdown() || executor.isTerminated())
{
skipped = true;
return ImmediateFuture.success(null);
}
return executor.submit(task);
}
catch (RejectedExecutionException ex)
{
skipped = true;
return ImmediateFuture.success(null);
}
else
finally
{
return executor.submit(task, null);
if (skipped)
{
logger.info("Skipped scheduled pulling schema from other nodes: the MIGRATION executor service has been shutdown.");
}
}
}

3
test/distributed/org/apache/cassandra/distributed/action/GossipHelper.java
View File

@ -51,6 +51,7 @@ import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.distributed.impl.DistributedTestSnitch.toCassandraInetAddressAndPort;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.junit.Assert.assertTrue;
public class GossipHelper
{
@ -228,7 +229,7 @@ public class GossipHelper
InetAddressAndPort endpoint = toCassandraInetAddressAndPort(pullFrom);
EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(endpoint);
Gossiper.instance.doOnChangeNotifications(endpoint, ApplicationState.SCHEMA, state.getApplicationState(ApplicationState.SCHEMA));
Schema.instance.waitUntilReady(Duration.ofSeconds(10));
assertTrue("schema is ready", Schema.instance.waitUntilReady(Duration.ofSeconds(10)));
}).accept(pullFrom);
}
}

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

@ -35,7 +35,9 @@ import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Stream;
import javax.management.ListenerNotFoundException;
import javax.management.Notification;
@ -104,6 +106,7 @@ import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.NoPayload;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.schema.MigrationCoordinator;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.service.ActiveRepairService;
@ -159,7 +162,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
{
public final IInstanceConfig config;
private volatile boolean initialized = false;
private final long startedAt;
private final AtomicLong startedAt = new AtomicLong();
@Deprecated
Instance(IInstanceConfig config, ClassLoader classLoader)
@ -181,11 +184,6 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
Object clusterId = Objects.requireNonNull(config.get(Constants.KEY_DTEST_API_CLUSTER_ID), "cluster_id is not defined");
ClusterIDDefiner.setId("cluster-" + clusterId);
InstanceIDDefiner.setInstanceId(config.num());
// Defer initialisation of Clock.Global until cluster/instance identifiers are set.
// Otherwise, the instance classloader's logging classes are setup ahead of time and
// the patterns/file paths are not set correctly. This will be addressed in a subsequent
// commit to extend the functionality of the @Shared annotation to app classes.
startedAt = nanoTime();
FBUtilities.setBroadcastInetAddressAndPort(InetAddressAndPort.getByAddressOverrideDefaults(config.broadcastAddress().getAddress(),
config.broadcastAddress().getPort()));
@ -538,6 +536,12 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
@Override
public void startup(ICluster cluster)
{
// Defer initialisation of Clock.Global until cluster/instance identifiers are set.
// Otherwise, the instance classloader's logging classes are setup ahead of time and
// the patterns/file paths are not set correctly. This will be addressed in a subsequent
// commit to extend the functionality of the @Shared annotation to app classes.
assert startedAt.compareAndSet(0L, System.nanoTime()) : "startedAt uninitialized";
sync(() -> {
try
{
@ -631,6 +635,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
StorageService.instance.registerDaemon(CassandraDaemon.getInstanceForTesting());
if (config.has(GOSSIP))
{
MigrationCoordinator.setUptimeFn(() -> TimeUnit.NANOSECONDS.toMillis(nanoTime() - startedAt.get()));
try
{
StorageService.instance.initServer();
@ -812,6 +817,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
finally
{
super.shutdown();
startedAt.set(0L);
}
});
}

12
test/distributed/org/apache/cassandra/distributed/test/MigrationCoordinatorTest.java
View File

@ -28,9 +28,10 @@ import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.IInstanceConfig;
import org.apache.cassandra.distributed.api.TokenSupplier;
import org.apache.cassandra.distributed.shared.NetworkTopology;
import org.apache.cassandra.schema.MigrationCoordinator;
import org.apache.cassandra.schema.Schema;
import static org.apache.cassandra.config.CassandraRelevantProperties.IGNORED_SCHEMA_CHECK_ENDPOINTS;
import static org.apache.cassandra.config.CassandraRelevantProperties.IGNORED_SCHEMA_CHECK_VERSIONS;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
@ -42,8 +43,9 @@ public class MigrationCoordinatorTest extends TestBaseImpl
{
System.clearProperty("cassandra.replace_address");
System.clearProperty("cassandra.consistent.rangemovement");
System.clearProperty(MigrationCoordinator.IGNORED_ENDPOINTS_PROP);
System.clearProperty(MigrationCoordinator.IGNORED_VERSIONS_PROP);
System.clearProperty(IGNORED_SCHEMA_CHECK_VERSIONS.getKey());
System.clearProperty(IGNORED_SCHEMA_CHECK_VERSIONS.getKey());
}
/**
* We shouldn't wait on versions only available from a node being replaced
@ -86,7 +88,7 @@ public class MigrationCoordinatorTest extends TestBaseImpl
IInstanceConfig config = cluster.newInstanceConfig();
config.set("auto_bootstrap", true);
System.setProperty(MigrationCoordinator.IGNORED_ENDPOINTS_PROP, ignoredEndpoint.getHostAddress());
IGNORED_SCHEMA_CHECK_ENDPOINTS.setString(ignoredEndpoint.getHostAddress());
System.setProperty("cassandra.consistent.rangemovement", "false");
cluster.bootstrap(config).startup();
}
@ -113,7 +115,7 @@ public class MigrationCoordinatorTest extends TestBaseImpl
IInstanceConfig config = cluster.newInstanceConfig();
config.set("auto_bootstrap", true);
System.setProperty(MigrationCoordinator.IGNORED_VERSIONS_PROP, initialVersion.toString() + ',' + oldVersion.toString());
IGNORED_SCHEMA_CHECK_VERSIONS.setString(initialVersion.toString() + ',' + oldVersion);
System.setProperty("cassandra.consistent.rangemovement", "false");
cluster.bootstrap(config).startup();
}

3
test/simulator/main/org/apache/cassandra/simulator/cluster/OnInstanceSyncSchemaForBootstrap.java
View File

@ -25,12 +25,13 @@ import org.apache.cassandra.simulator.systems.SimulatedActionTask;
import static org.apache.cassandra.simulator.Action.Modifier.DISPLAY_ORIGIN;
import static org.apache.cassandra.simulator.Action.Modifiers.RELIABLE_NO_TIMEOUTS;
import static org.junit.Assert.assertTrue;
class OnInstanceSyncSchemaForBootstrap extends SimulatedActionTask
{
public OnInstanceSyncSchemaForBootstrap(ClusterActions actions, int node)
{
super("Sync Schema on " + node, RELIABLE_NO_TIMEOUTS.with(DISPLAY_ORIGIN), RELIABLE_NO_TIMEOUTS, actions, actions.cluster.get(node),
() -> Schema.instance.waitUntilReady(Duration.ofMinutes(10)));
() -> assertTrue("schema is ready", Schema.instance.waitUntilReady(Duration.ofMinutes(10))));
}
}