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 fa85ff978bae303fe1b06dce64b758b635278a4d Merge: 860de83 0d5ccb9 Author: Alex Petrov <[email protected]> AuthorDate: Mon Nov 11 15:55:33 2019 +0100 Merge branch 'cassandra-3.11' into trunk .../org/apache/cassandra/gms/GossiperEvent.java | 4 +- .../apache/cassandra/service/CassandraDaemon.java | 47 ++++-- .../apache/cassandra/distributed/api/Feature.java | 2 +- .../cassandra/distributed/api/IInstance.java | 2 + .../apache/cassandra/distributed/api/IListen.java | 2 + .../distributed/impl/AbstractCluster.java | 158 +++++++++++++++++---- .../cassandra/distributed/impl/Instance.java | 25 +++- .../distributed/impl/InstanceClassLoader.java | 16 ++- .../cassandra/distributed/impl/InstanceConfig.java | 34 +++-- .../apache/cassandra/distributed/impl/Listen.java | 9 ++ .../apache/cassandra/distributed/impl/RowUtil.java | 17 +++ .../distributed/test/DistributedTestBase.java | 7 + .../distributed/test/NativeProtocolTest.java | 59 ++++++++ .../distributed/test/ResourceLeakTest.java | 16 +++ 14 files changed, 347 insertions(+), 51 deletions(-) diff --cc src/java/org/apache/cassandra/gms/GossiperEvent.java index 2de88bc,0000000..ef7bd8d mode 100644,000000..100644 --- a/src/java/org/apache/cassandra/gms/GossiperEvent.java +++ b/src/java/org/apache/cassandra/gms/GossiperEvent.java @@@ -1,111 -1,0 +1,111 @@@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.gms; + +import java.io.Serializable; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import javax.annotation.Nullable; + +import org.apache.cassandra.diag.DiagnosticEvent; +import org.apache.cassandra.locator.InetAddressAndPort; + +/** + * DiagnosticEvent implementation for {@link Gossiper} activities. + */ - final class GossiperEvent extends DiagnosticEvent ++public final class GossiperEvent extends DiagnosticEvent +{ + private final InetAddressAndPort endpoint; + @Nullable + private final Long quarantineExpiration; + @Nullable + private final EndpointState localState; + + private final Map<InetAddressAndPort, EndpointState> endpointStateMap; + private final boolean inShadowRound; + private final Map<InetAddressAndPort, Long> justRemovedEndpoints; + private final long lastProcessedMessageAt; + private final Set<InetAddressAndPort> liveEndpoints; + private final List<String> seeds; + private final Set<InetAddressAndPort> seedsInShadowRound; + private final Map<InetAddressAndPort, Long> unreachableEndpoints; + + - enum GossiperEventType ++ public enum GossiperEventType + { + MARKED_AS_SHUTDOWN, + CONVICTED, + REPLACEMENT_QUARANTINE, + REPLACED_ENDPOINT, + EVICTED_FROM_MEMBERSHIP, + REMOVED_ENDPOINT, + QUARANTINED_ENDPOINT, + MARKED_ALIVE, + REAL_MARKED_ALIVE, + MARKED_DEAD, + MAJOR_STATE_CHANGE_HANDLED, + SEND_GOSSIP_DIGEST_SYN + } + + public GossiperEventType type; + + + GossiperEvent(GossiperEventType type, Gossiper gossiper, InetAddressAndPort endpoint, + @Nullable Long quarantineExpiration, @Nullable EndpointState localState) + { + this.type = type; + this.endpoint = endpoint; + this.quarantineExpiration = quarantineExpiration; + this.localState = localState; + + this.endpointStateMap = gossiper.getEndpointStateMap(); + this.inShadowRound = gossiper.isInShadowRound(); + this.justRemovedEndpoints = gossiper.getJustRemovedEndpoints(); + this.lastProcessedMessageAt = gossiper.getLastProcessedMessageAt(); + this.liveEndpoints = gossiper.getLiveMembers(); + this.seeds = gossiper.getSeeds(); + this.seedsInShadowRound = gossiper.getSeedsInShadowRound(); + this.unreachableEndpoints = gossiper.getUnreachableEndpoints(); + } + + public Enum<GossiperEventType> getType() + { + return type; + } + + public HashMap<String, Serializable> toMap() + { + // be extra defensive against nulls and bugs + HashMap<String, Serializable> ret = new HashMap<>(); + if (endpoint != null) ret.put("endpoint", endpoint.getHostAddress(true)); + ret.put("quarantineExpiration", quarantineExpiration); + ret.put("localState", String.valueOf(localState)); + ret.put("endpointStateMap", String.valueOf(endpointStateMap)); + ret.put("inShadowRound", inShadowRound); + ret.put("justRemovedEndpoints", String.valueOf(justRemovedEndpoints)); + ret.put("lastProcessedMessageAt", lastProcessedMessageAt); + ret.put("liveEndpoints", String.valueOf(liveEndpoints)); + ret.put("seeds", String.valueOf(seeds)); + ret.put("seedsInShadowRound", String.valueOf(seedsInShadowRound)); + ret.put("unreachableEndpoints", String.valueOf(unreachableEndpoints)); + return ret; + } +} diff --cc src/java/org/apache/cassandra/service/CassandraDaemon.java index 9d05371,16a6145..d705bd7 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@@ -252,19 -264,8 +258,18 @@@ public class CassandraDaemo // Populate token metadata before flushing, for token-aware sstable partitioning (#6696) StorageService.instance.populateTokenMetadata(); - // load schema from disk - Schema.instance.loadFromDisk(); + try + { + // load schema from disk + Schema.instance.loadFromDisk(); + } + catch (Exception e) + { + logger.error("Error while loading schema: ", e); + throw e; + } + - VirtualKeyspaceRegistry.instance.register(VirtualSchemaKeyspace.instance); - VirtualKeyspaceRegistry.instance.register(SystemViewsKeyspace.instance); ++ setupVirtualKeyspaces(); // clean up debris in the rest of the keyspaces for (String keyspaceName : Schema.instance.getKeyspaces()) @@@ -430,20 -434,22 +435,31 @@@ // due to scheduling errors or race conditions ScheduledExecutors.optionalTasks.scheduleWithFixedDelay(ColumnFamilyStore.getBackgroundCompactionTaskSubmitter(), 5, 1, TimeUnit.MINUTES); - // Thrift - InetAddress rpcAddr = DatabaseDescriptor.getRpcAddress(); - int rpcPort = DatabaseDescriptor.getRpcPort(); - int listenBacklog = DatabaseDescriptor.getRpcListenBacklog(); - thriftServer = new ThriftServer(rpcAddr, rpcPort, listenBacklog); + // schedule periodic recomputation of speculative retry thresholds + ScheduledExecutors.optionalTasks.scheduleWithFixedDelay( + () -> Keyspace.all().forEach(k -> k.getColumnFamilyStores().forEach(ColumnFamilyStore::updateSpeculationThreshold)), + DatabaseDescriptor.getReadRpcTimeout(NANOSECONDS), + DatabaseDescriptor.getReadRpcTimeout(NANOSECONDS), + NANOSECONDS + ); + - // Native transport - nativeTransportService = new NativeTransportService(); + initializeNativeTransport(); completeSetup(); } ++ public void setupVirtualKeyspaces() ++ { ++ VirtualKeyspaceRegistry.instance.register(VirtualSchemaKeyspace.instance); ++ VirtualKeyspaceRegistry.instance.register(SystemViewsKeyspace.instance); ++ } ++ + public void initializeNativeTransport() + { + // Native transport + nativeTransportService = new NativeTransportService(); + } + /* * Asynchronously load the row and key cache in one off threads and return a compound future of the result. * Error handling is pushed into the cache load since cache loads are allowed to fail and are handled by logging. diff --cc test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java index d866fae,6d24ad9..28a6a74 --- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java @@@ -129,9 -131,9 +131,9 @@@ public abstract class AbstractCluster< private IInvokableInstance newInstance(int generation) { - ClassLoader classLoader = new InstanceClassLoader(generation, version.classpath, sharedClassLoader); + ClassLoader classLoader = new InstanceClassLoader(generation, config.num, version.classpath, sharedClassLoader); return Instance.transferAdhoc((SerializableBiFunction<IInstanceConfig, ClassLoader, Instance>)Instance::new, classLoader) - .apply(config, classLoader); + .apply(config.forVersion(version.major), classLoader); } public IInstanceConfig config() @@@ -256,12 -273,11 +273,11 @@@ FBUtilities.waitOnFutures(instances.stream() .map(i -> i.async(consumer).apply(i)) .collect(Collectors.toList()), - timeout, units); + timeout, unit); } - public IMessageFilters filters() { return filters; } - public MessageFilters.Builder verbs(MessagingService.Verb ... verbs) { return filters.verbs(verbs); } + public MessageFilters.Builder verbs(Verb... verbs) { return filters.verbs(verbs); } public void disableAutoCompaction(String keyspace) { @@@ -358,9 -438,27 +438,27 @@@ get(instance).schemaChangeInternal(statement); } - public void startup() + void startup() { - parallelForEach(I::startup, 0, null); + try (AllMembersAliveMonitor monitor = new AllMembersAliveMonitor()) + { + // 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. + List<I> startSequentially = new ArrayList<>(); + List<I> startParallel = new ArrayList<>(); + for (I instance : instances) + { + if ((boolean) instance.config().get("auto_bootstrap")) + startSequentially.add(instance); + else + startParallel.add(instance); + } + + forEach(startSequentially, I::startup); + parallelForEach(startParallel, I::startup, 0, null); + monitor.waitForCompletion(); + } } protected interface Factory<I extends IInstance, C extends AbstractCluster<I>> diff --cc test/distributed/org/apache/cassandra/distributed/impl/Instance.java index faf59d5,6c3a70d..cdc3cc8 --- a/test/distributed/org/apache/cassandra/distributed/impl/Instance.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Instance.java @@@ -68,11 -72,12 +68,12 @@@ import org.apache.cassandra.io.sstable. import org.apache.cassandra.io.util.DataInputBuffer; import org.apache.cassandra.io.util.DataOutputBuffer; import org.apache.cassandra.locator.InetAddressAndPort; -import org.apache.cassandra.net.IMessageSink; -import org.apache.cassandra.net.MessageIn; -import org.apache.cassandra.net.MessageOut; +import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; -import org.apache.cassandra.schema.LegacySchemaMigrator; +import org.apache.cassandra.schema.Schema; +import org.apache.cassandra.schema.SchemaConstants; +import org.apache.cassandra.service.ActiveRepairService; + import org.apache.cassandra.service.CassandraDaemon; import org.apache.cassandra.service.ClientState; import org.apache.cassandra.service.PendingRangeCalculatorService; import org.apache.cassandra.service.QueryState; @@@ -367,7 -432,15 +370,16 @@@ public class Instance extends IsolatedE SystemKeyspace.finishStartup(); + if (config.has(NATIVE_PROTOCOL)) + { ++ // Start up virtual table support ++ CassandraDaemon.getInstanceForTesting().setupVirtualKeyspaces(); ++ + CassandraDaemon.getInstanceForTesting().initializeNativeTransport(); + CassandraDaemon.getInstanceForTesting().startNativeTransport(); + } + - if (!FBUtilities.getBroadcastAddress().equals(broadcastAddressAndPort().address)) - throw new IllegalStateException(); - if (DatabaseDescriptor.getStoragePort() != broadcastAddressAndPort().port) + if (!FBUtilities.getBroadcastAddressAndPort().equals(broadcastAddressAndPort())) throw new IllegalStateException(); } catch (Throwable t) @@@ -507,11 -578,15 +521,20 @@@ .thenRun(super::shutdown); } + public int liveMemberCount() + { + return sync(() -> { + if (!DatabaseDescriptor.isDaemonInitialized() || !Gossiper.instance.isEnabled()) + return 0; + return Gossiper.instance.getLiveMembers().size(); + }).call(); + } + + private static void shutdownAndWait(List<ExecutorService> executors) throws TimeoutException, InterruptedException + { + ExecutorUtils.shutdownNow(executors); + ExecutorUtils.awaitTermination(1L, MINUTES, executors); + } private static Throwable parallelRun(Throwable accumulate, ExecutorService runOn, ThrowingRunnable ... runnables) { diff --cc test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java index 3b54578,d36487e..fcedf21 --- a/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java @@@ -108,9 -114,12 +114,14 @@@ public class InstanceConfig implements .set("storage_port", 7012) .set("endpoint_snitch", DistributedTestSnitch.class.getName()) .set("seed_provider", new ParameterizedClass(SimpleSeedProvider.class.getName(), - Collections.singletonMap("seeds", "127.0.0.1"))) + Collections.singletonMap("seeds", "127.0.0.1:7012"))) + // required settings for dtest functionality + .set("diagnostic_events_enabled", true) + .set("auto_bootstrap", false) + // capacities that are based on `totalMemory` that should be fixed size + .set("index_summary_capacity_in_mb", 50l) + .set("counter_cache_size_in_mb", 50l) + .set("key_cache_size_in_mb", 50l) // legacy parameters .forceSet("commitlog_sync_batch_window_in_ms", 1.0); this.featureFlags = EnumSet.noneOf(Feature.class); diff --cc test/distributed/org/apache/cassandra/distributed/impl/Listen.java index ec34518,27ae156..e37c4f7 --- a/test/distributed/org/apache/cassandra/distributed/impl/Listen.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/Listen.java @@@ -18,11 -18,13 +18,12 @@@ package org.apache.cassandra.distributed.impl; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.locks.LockSupport; -import java.util.function.Supplier; +import java.util.function.Consumer; +import org.apache.cassandra.diag.DiagnosticEventService; import org.apache.cassandra.distributed.api.IListen; -import org.apache.cassandra.gms.Gossiper; ++import org.apache.cassandra.gms.GossiperEvent; +import org.apache.cassandra.schema.SchemaEvent; public class Listen implements IListen { @@@ -34,8 -37,31 +36,15 @@@ public Cancel schema(Runnable onChange) { - return start(onChange, instance::schemaVersion); + Consumer<SchemaEvent> consumer = event -> onChange.run(); + DiagnosticEventService.instance().subscribe(SchemaEvent.class, SchemaEvent.SchemaEventType.VERSION_UPDATED, consumer); + return () -> DiagnosticEventService.instance().unsubscribe(SchemaEvent.class, consumer); } + + public Cancel liveMembers(Runnable onChange) + { - return start(onChange, instance::liveMemberCount); - } - - protected <T> Cancel start(Runnable onChange, Supplier<T> valueSupplier) { - AtomicBoolean cancel = new AtomicBoolean(false); - instance.isolatedExecutor.execute(() -> { - T prev = valueSupplier.get(); - while (true) - { - if (cancel.get()) - return; - - LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10L)); - - T cur = valueSupplier.get(); - if (!prev.equals(cur)) - onChange.run(); - prev = cur; - } - }); - return () -> cancel.set(true); ++ Consumer<GossiperEvent> consumer = event -> onChange.run(); ++ DiagnosticEventService.instance().subscribe(GossiperEvent.class, GossiperEvent.GossiperEventType.REAL_MARKED_ALIVE, consumer); ++ return () -> DiagnosticEventService.instance().unsubscribe(GossiperEvent.class, consumer); + } } diff --cc test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java index 35d3691,03e4047..d048079 --- a/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/DistributedTestBase.java @@@ -78,9 -80,14 +80,14 @@@ public class DistributedTestBas return cluster; } + public static void assertRows(ResultSet actual,Object[]... expected) + { + assertRows(RowUtil.toObjects(actual), expected); + } + public static void assertRows(Object[][] actual, Object[]... expected) { - Assert.assertEquals(rowsNotEqualErrorMessage(actual, expected), + Assert.assertEquals(rowsNotEqualErrorMessage(expected, actual), expected.length, actual.length); for (int i = 0; i < expected.length; i++) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
