| /* |
| * 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.distributed.impl; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.net.InetAddress; |
| import java.net.InetSocketAddress; |
| import java.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.UUID; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.CopyOnWriteArrayList; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.TimeoutException; |
| import java.util.function.BiConsumer; |
| import java.util.function.Function; |
| |
| import javax.management.ListenerNotFoundException; |
| import javax.management.Notification; |
| import javax.management.NotificationListener; |
| |
| import org.apache.cassandra.batchlog.BatchlogManager; |
| import org.apache.cassandra.concurrent.ScheduledExecutors; |
| import org.apache.cassandra.concurrent.SharedExecutorPool; |
| import org.apache.cassandra.concurrent.StageManager; |
| import org.apache.cassandra.config.Config; |
| import org.apache.cassandra.config.DatabaseDescriptor; |
| import org.apache.cassandra.config.Schema; |
| import org.apache.cassandra.cql3.CQLStatement; |
| import org.apache.cassandra.cql3.QueryOptions; |
| import org.apache.cassandra.cql3.QueryProcessor; |
| import org.apache.cassandra.cql3.statements.ParsedStatement; |
| import org.apache.cassandra.db.ColumnFamilyStore; |
| import org.apache.cassandra.db.Keyspace; |
| import org.apache.cassandra.db.Memtable; |
| import org.apache.cassandra.db.SystemKeyspace; |
| import org.apache.cassandra.db.commitlog.CommitLog; |
| import org.apache.cassandra.db.compaction.CompactionManager; |
| import org.apache.cassandra.dht.IPartitioner; |
| import org.apache.cassandra.dht.Token; |
| import org.apache.cassandra.distributed.api.ICluster; |
| import org.apache.cassandra.distributed.api.ICoordinator; |
| import org.apache.cassandra.distributed.api.IInstance; |
| import org.apache.cassandra.distributed.api.IInstanceConfig; |
| 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.NodeToolResult; |
| import org.apache.cassandra.distributed.api.SimpleQueryResult; |
| import org.apache.cassandra.distributed.mock.nodetool.InternalNodeProbe; |
| import org.apache.cassandra.distributed.mock.nodetool.InternalNodeProbeFactory; |
| import org.apache.cassandra.distributed.shared.NetworkTopology; |
| import org.apache.cassandra.gms.ApplicationState; |
| import org.apache.cassandra.gms.Gossiper; |
| import org.apache.cassandra.gms.VersionedValue; |
| import org.apache.cassandra.hints.HintsService; |
| import org.apache.cassandra.index.SecondaryIndexManager; |
| import org.apache.cassandra.io.IVersionedSerializer; |
| import org.apache.cassandra.io.sstable.IndexSummaryManager; |
| import org.apache.cassandra.io.sstable.format.SSTableReader; |
| import org.apache.cassandra.io.util.DataInputBuffer; |
| import org.apache.cassandra.io.util.DataOutputBuffer; |
| import org.apache.cassandra.io.util.DataOutputPlus; |
| import org.apache.cassandra.net.CompactEndpointSerializationHelper; |
| import org.apache.cassandra.net.IMessageSink; |
| import org.apache.cassandra.net.MessageIn; |
| import org.apache.cassandra.net.MessageOut; |
| import org.apache.cassandra.net.MessagingService; |
| import org.apache.cassandra.schema.LegacySchemaMigrator; |
| import org.apache.cassandra.service.CassandraDaemon; |
| import org.apache.cassandra.service.ClientState; |
| import org.apache.cassandra.service.PendingRangeCalculatorService; |
| import org.apache.cassandra.service.QueryState; |
| import org.apache.cassandra.service.StorageService; |
| import org.apache.cassandra.service.StorageServiceMBean; |
| import org.apache.cassandra.streaming.StreamCoordinator; |
| import org.apache.cassandra.tools.NodeTool; |
| import org.apache.cassandra.tracing.TraceState; |
| import org.apache.cassandra.tracing.Tracing; |
| import org.apache.cassandra.transport.messages.ResultMessage; |
| import org.apache.cassandra.utils.DiagnosticSnapshotService; |
| import org.apache.cassandra.utils.ExecutorUtils; |
| import org.apache.cassandra.utils.FBUtilities; |
| import org.apache.cassandra.utils.NanoTimeToCurrentTimeMillis; |
| import org.apache.cassandra.utils.Pair; |
| import org.apache.cassandra.utils.Throwables; |
| import org.apache.cassandra.utils.UUIDGen; |
| import org.apache.cassandra.utils.concurrent.Ref; |
| import org.apache.cassandra.utils.memory.BufferPool; |
| |
| import static java.util.concurrent.TimeUnit.MINUTES; |
| import static org.apache.cassandra.distributed.api.Feature.GOSSIP; |
| import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL; |
| import static org.apache.cassandra.distributed.api.Feature.NETWORK; |
| |
| public class Instance extends IsolatedExecutor implements IInvokableInstance |
| { |
| private static final Map<Class<?>, Function<Object, Object>> mapper = new HashMap<Class<?>, Function<Object, Object>>() {{ |
| this.put(IInstanceConfig.ParameterizedClass.class, (obj) -> { |
| IInstanceConfig.ParameterizedClass pc = (IInstanceConfig.ParameterizedClass) obj; |
| return new org.apache.cassandra.config.ParameterizedClass(pc.class_name, pc.parameters); |
| }); |
| }}; |
| |
| public final IInstanceConfig config; |
| |
| // should never be invoked directly, so that it is instantiated on other class loader; |
| // only visible for inheritance |
| Instance(IInstanceConfig config, ClassLoader classLoader) |
| { |
| super("node" + config.num(), classLoader); |
| this.config = config; |
| InstanceIDDefiner.setInstanceId(config.num()); |
| FBUtilities.setBroadcastInetAddress(config.broadcastAddress().getAddress()); |
| |
| // Set the config at instance creation, possibly before startup() has run on all other instances. |
| // setMessagingVersions below will call runOnInstance which will instantiate |
| // the MessagingService and dependencies preventing later changes to network parameters. |
| Config.setOverrideLoadConfig(() -> loadConfig(config)); |
| } |
| |
| @Override |
| public IInstanceConfig config() |
| { |
| return config; |
| } |
| |
| @Override |
| public ICoordinator coordinator() |
| { |
| return new Coordinator(this); |
| } |
| |
| public IListen listen() |
| { |
| return new Listen(this); |
| } |
| |
| @Override |
| public InetSocketAddress broadcastAddress() { return config.broadcastAddress(); } |
| |
| @Override |
| public SimpleQueryResult executeInternalWithResult(String query, Object... args) |
| { |
| return sync(() -> { |
| ParsedStatement.Prepared prepared = QueryProcessor.prepareInternal(query); |
| ResultMessage result = prepared.statement.executeInternal(QueryProcessor.internalQueryState(), |
| QueryProcessor.makeInternalOptions(prepared, args)); |
| |
| return RowUtil.toQueryResult(result); |
| }).call(); |
| } |
| |
| @Override |
| public UUID schemaVersion() |
| { |
| // we do not use method reference syntax here, because we need to sync on the node-local schema instance |
| //noinspection Convert2MethodRef |
| return Schema.instance.getVersion(); |
| } |
| |
| public void startup() |
| { |
| throw new UnsupportedOperationException(); |
| } |
| |
| public boolean isShutdown() |
| { |
| return isolatedExecutor.isShutdown(); |
| } |
| |
| @Override |
| public void schemaChangeInternal(String query) |
| { |
| sync(() -> { |
| try |
| { |
| ClientState state = ClientState.forInternalCalls(); |
| state.setKeyspace(SystemKeyspace.NAME); |
| QueryState queryState = new QueryState(state); |
| |
| CQLStatement statement = QueryProcessor.parseStatement(query, queryState).statement; |
| statement.validate(state); |
| |
| QueryOptions options = QueryOptions.forInternalCalls(Collections.emptyList()); |
| statement.executeInternal(queryState, options); |
| } |
| catch (Exception e) |
| { |
| throw new RuntimeException("Error setting schema for test (query was: " + query + ")", e); |
| } |
| }).run(); |
| } |
| |
| private void registerMockMessaging(ICluster<IInstance> cluster) |
| { |
| BiConsumer<InetSocketAddress, IMessage> deliverToInstance = (to, message) -> cluster.get(to).receiveMessage(message); |
| BiConsumer<InetSocketAddress, IMessage> deliverToInstanceIfNotFiltered = (to, message) -> { |
| int fromNum = config().num(); |
| int toNum = cluster.get(to).config().num(); |
| |
| if (cluster.filters().permitOutbound(fromNum, toNum, message) |
| && cluster.filters().permitInbound(fromNum, toNum, message)) |
| deliverToInstance.accept(to, message); |
| }; |
| |
| Map<InetAddress, InetSocketAddress> addressAndPortMap = new HashMap<>(); |
| cluster.stream().forEach(instance -> { |
| InetSocketAddress addressAndPort = instance.broadcastAddress(); |
| if (!addressAndPort.equals(instance.config().broadcastAddress())) |
| throw new IllegalStateException("addressAndPort mismatch: " + addressAndPort + " vs " + instance.config().broadcastAddress()); |
| InetSocketAddress prev = addressAndPortMap.put(addressAndPort.getAddress(), |
| addressAndPort); |
| if (null != prev) |
| throw new IllegalStateException("This version of Cassandra does not support multiple nodes with the same InetAddress: " + addressAndPort + " vs " + prev); |
| }); |
| |
| MessagingService.instance().addMessageSink(new MessageDeliverySink(deliverToInstanceIfNotFiltered, addressAndPortMap::get)); |
| } |
| |
| // unnecessary if registerMockMessaging used |
| private void registerFilters(ICluster cluster) |
| { |
| IInstance instance = this; |
| MessagingService.instance().addMessageSink(new IMessageSink() |
| { |
| public boolean allowOutgoingMessage(MessageOut message, int id, InetAddress toAddress) |
| { |
| // Port is not passed in, so take a best guess at the destination port from this instance |
| IInstance to = cluster.get(NetworkTopology.addressAndPort(toAddress, |
| instance.config().broadcastAddress().getPort())); |
| int fromNum = config().num(); |
| int toNum = to.config().num(); |
| return cluster.filters().permitOutbound(fromNum, toNum, serializeMessage(message, id, |
| broadcastAddress(), |
| to.config().broadcastAddress())); |
| } |
| |
| public boolean allowIncomingMessage(MessageIn message, int id) |
| { |
| // Port is not passed in, so take a best guess at the destination port from this instance |
| IInstance from = cluster.get(NetworkTopology.addressAndPort(message.from, |
| instance.config().broadcastAddress().getPort())); |
| int fromNum = from.config().num(); |
| int toNum = config().num(); |
| |
| |
| IMessage msg = serializeMessage(message, id, from.broadcastAddress(), broadcastAddress()); |
| |
| return cluster.filters().permitInbound(fromNum, toNum, msg); |
| } |
| }); |
| } |
| |
| public static IMessage serializeMessage(MessageOut messageOut, int id, InetSocketAddress from, InetSocketAddress to) |
| { |
| try (DataOutputBuffer out = new DataOutputBuffer(1024)) |
| { |
| int version = MessagingService.instance().getVersion(to.getAddress()); |
| |
| out.writeInt(MessagingService.PROTOCOL_MAGIC); |
| out.writeInt(id); |
| long timestamp = System.currentTimeMillis(); |
| out.writeInt((int) timestamp); |
| messageOut.serialize(out, version); |
| return new MessageImpl(messageOut.verb.ordinal(), out.toByteArray(), id, version, from); |
| } |
| catch (IOException e) |
| { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| public static IMessage serializeMessage(MessageIn<?> messageIn, int id, InetSocketAddress from, InetSocketAddress to) |
| { |
| try (DataOutputBuffer out = new DataOutputBuffer(1024)) |
| { |
| // Serialize header |
| int version = MessagingService.instance().getVersion(to.getAddress()); |
| |
| out.writeInt(MessagingService.PROTOCOL_MAGIC); |
| out.writeInt(id); |
| long timestamp = System.currentTimeMillis(); |
| out.writeInt((int) timestamp); |
| |
| // Serialize the message itself |
| IVersionedSerializer serializer = MessagingService.instance().verbSerializers.get(messageIn.verb); |
| CompactEndpointSerializationHelper.serialize(from.getAddress(), out); |
| |
| out.writeInt(MessagingService.Verb.convertForMessagingServiceVersion(messageIn.verb, version).ordinal()); |
| out.writeInt(messageIn.parameters.size()); |
| for (Map.Entry<String, byte[]> entry : messageIn.parameters.entrySet()) |
| { |
| out.writeUTF(entry.getKey()); |
| out.writeInt(entry.getValue().length); |
| out.write(entry.getValue()); |
| } |
| |
| if (messageIn.payload != null && serializer != MessagingService.CallbackDeterminedSerializer.instance) |
| { |
| try (DataOutputBuffer dob = new DataOutputBuffer()) |
| { |
| serializer.serialize(messageIn.payload, dob, version); |
| |
| int size = dob.getLength(); |
| out.writeInt(size); |
| out.write(dob.getData(), 0, size); |
| } |
| } |
| else |
| { |
| out.writeInt(0); |
| } |
| |
| |
| return new MessageImpl(messageIn.verb.ordinal(), out.toByteArray(), id, version, from); |
| } |
| catch (IOException e) |
| { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| private class MessageDeliverySink implements IMessageSink |
| { |
| private final BiConsumer<InetSocketAddress, IMessage> deliver; |
| private final Function<InetAddress, InetSocketAddress> lookupAddressAndPort; |
| |
| MessageDeliverySink(BiConsumer<InetSocketAddress, IMessage> deliver, |
| Function<InetAddress, InetSocketAddress> lookupAddressAndPort) |
| { |
| this.deliver = deliver; |
| this.lookupAddressAndPort = lookupAddressAndPort; |
| } |
| |
| public boolean allowOutgoingMessage(MessageOut messageOut, int id, InetAddress to) |
| { |
| InetSocketAddress from = broadcastAddress(); |
| assert from.equals(lookupAddressAndPort.apply(messageOut.from)); |
| |
| // Tracing logic - similar to org.apache.cassandra.net.OutboundTcpConnection.writeConnected |
| byte[] sessionBytes = (byte[]) messageOut.parameters.get(Tracing.TRACE_HEADER); |
| if (sessionBytes != null) |
| { |
| UUID sessionId = UUIDGen.getUUID(ByteBuffer.wrap(sessionBytes)); |
| TraceState state = Tracing.instance.get(sessionId); |
| String traceMessage = String.format("Sending %s message to %s", messageOut.verb, to); |
| // session may have already finished; see CASSANDRA-5668 |
| if (state == null) |
| { |
| byte[] traceTypeBytes = (byte[]) messageOut.parameters.get(Tracing.TRACE_TYPE); |
| Tracing.TraceType traceType = traceTypeBytes == null ? Tracing.TraceType.QUERY : Tracing.TraceType.deserialize(traceTypeBytes[0]); |
| TraceState.mutateWithTracing(ByteBuffer.wrap(sessionBytes), traceMessage, -1, traceType.getTTL()); |
| } |
| else |
| { |
| state.trace(traceMessage); |
| if (messageOut.verb == MessagingService.Verb.REQUEST_RESPONSE) |
| Tracing.instance.doneWithNonLocalSession(state); |
| } |
| } |
| |
| InetSocketAddress toFull = lookupAddressAndPort.apply(to); |
| deliver.accept(toFull, |
| serializeMessage(messageOut, id, broadcastAddress(), toFull)); |
| |
| return false; |
| } |
| |
| public boolean allowIncomingMessage(MessageIn message, int id) |
| { |
| // we can filter to our heart's content on the outgoing message; no need to worry about incoming |
| return true; |
| } |
| } |
| |
| public static Pair<MessageIn<Object>, Integer> deserializeMessage(IMessage imessage) |
| { |
| // Based on org.apache.cassandra.net.IncomingTcpConnection.receiveMessage |
| try (DataInputBuffer input = new DataInputBuffer(imessage.bytes())) |
| { |
| int version = imessage.version(); |
| if (version > MessagingService.current_version) |
| { |
| throw new IllegalStateException(String.format("Received message version %d but current version is %d", |
| version, |
| MessagingService.current_version)); |
| } |
| |
| MessagingService.validateMagic(input.readInt()); |
| int id; |
| if (version < MessagingService.VERSION_20) |
| id = Integer.parseInt(input.readUTF()); |
| else |
| id = input.readInt(); |
| if (imessage.id() != id) |
| throw new IllegalStateException(String.format("Message id mismatch: %d != %d", imessage.id(), id)); |
| |
| // make sure to readInt, even if cross_node_to is not enabled |
| int partial = input.readInt(); |
| |
| return Pair.create(MessageIn.read(input, version, id), partial); |
| //long currentTime = ApproximateTime.currentTimeMillis(); |
| //return MessageIn.read(input, version, id, MessageIn.readConstructionTime(imessage.from().getAddress(), input, currentTime)); |
| } |
| catch (IOException e) |
| { |
| throw new RuntimeException(); |
| } |
| } |
| |
| 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); |
| } |
| |
| MessageIn<Object> message = deserialized.left; |
| int partial = deserialized.right; |
| |
| long timestamp = System.currentTimeMillis(); |
| boolean isCrossNodeTimestamp = false; |
| |
| 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); |
| } |
| }).run(); |
| } |
| |
| public int getMessagingVersion() |
| { |
| return callsOnInstance(() -> MessagingService.current_version).call(); |
| } |
| |
| public void setMessagingVersion(InetSocketAddress endpoint, int version) |
| { |
| runOnInstance(() -> MessagingService.instance().setVersion(endpoint.getAddress(), version)); |
| } |
| |
| public void flush(String keyspace) |
| { |
| runOnInstance(() -> FBUtilities.waitOnFutures(Keyspace.open(keyspace).flush())); |
| } |
| |
| public void forceCompact(String keyspace, String table) |
| { |
| runOnInstance(() -> { |
| try |
| { |
| Keyspace.open(keyspace).getColumnFamilyStore(table).forceMajorCompaction(); |
| } |
| catch (Exception e) |
| { |
| throw new RuntimeException(e); |
| } |
| }); |
| } |
| |
| @Override |
| public void startup(ICluster cluster) |
| { |
| sync(() -> { |
| try |
| { |
| mkdirs(); |
| |
| assert config.networkTopology().contains(config.broadcastAddress()); |
| DistributedTestSnitch.assign(config.networkTopology()); |
| |
| DatabaseDescriptor.setDaemonInitialized(); |
| DatabaseDescriptor.createAllDirectories(); |
| |
| // We need to persist this as soon as possible after startup checks. |
| // This should be the first write to SystemKeyspace (CASSANDRA-11742) |
| SystemKeyspace.persistLocalMetadata(); |
| LegacySchemaMigrator.migrate(); |
| |
| try |
| { |
| // load schema from disk |
| Schema.instance.loadFromDisk(); |
| } |
| catch (Exception e) |
| { |
| throw e; |
| } |
| |
| Keyspace.setInitialized(); |
| |
| // Replay any CommitLogSegments found on disk |
| try |
| { |
| CommitLog.instance.recover(); |
| } |
| catch (IOException e) |
| { |
| throw new RuntimeException(e); |
| } |
| |
| if (config.has(NETWORK)) |
| { |
| registerFilters(cluster); |
| MessagingService.instance().listen(); |
| } |
| else |
| { |
| // Even though we don't use MessagingService, access the static SocketFactory |
| // instance here so that we start the static event loop state |
| // -- not sure what that means? SocketFactory.instance.getClass(); |
| registerMockMessaging(cluster); |
| } |
| |
| // TODO: this is more than just gossip |
| if (config.has(GOSSIP)) |
| { |
| StorageService.instance.initServer(); |
| StorageService.instance.removeShutdownHook(); |
| } |
| else |
| { |
| initializeRing(cluster); |
| } |
| |
| StorageService.instance.ensureTraceKeyspace(); |
| |
| SystemKeyspace.finishStartup(); |
| |
| if (config.has(NATIVE_PROTOCOL)) |
| { |
| CassandraDaemon.getInstanceForTesting().initializeNativeTransport(); |
| CassandraDaemon.getInstanceForTesting().startNativeTransport(); |
| StorageService.instance.setRpcReady(true); |
| } |
| |
| if (!FBUtilities.getBroadcastAddress().equals(broadcastAddress().getAddress())) |
| throw new IllegalStateException(); |
| if (DatabaseDescriptor.getStoragePort() != broadcastAddress().getPort()) |
| throw new IllegalStateException(); |
| } |
| catch (Throwable t) |
| { |
| if (t instanceof RuntimeException) |
| throw (RuntimeException) t; |
| throw new RuntimeException(t); |
| } |
| }).run(); |
| } |
| |
| private void mkdirs() |
| { |
| new File(config.getString("saved_caches_directory")).mkdirs(); |
| new File(config.getString("hints_directory")).mkdirs(); |
| new File(config.getString("commitlog_directory")).mkdirs(); |
| for (String dir : (String[]) config.get("data_file_directories")) |
| new File(dir).mkdirs(); |
| } |
| |
| private static Config loadConfig(IInstanceConfig overrides) |
| { |
| Config config = new Config(); |
| overrides.propagate(config, mapper); |
| return config; |
| } |
| |
| private void initializeRing(ICluster cluster) |
| { |
| // This should be done outside instance in order to avoid serializing config |
| String partitionerName = config.getString("partitioner"); |
| List<String> initialTokens = new ArrayList<>(); |
| List<InetSocketAddress> hosts = new ArrayList<>(); |
| List<UUID> hostIds = new ArrayList<>(); |
| for (int i = 1 ; i <= cluster.size() ; ++i) |
| { |
| IInstanceConfig config = cluster.get(i).config(); |
| initialTokens.add(config.getString("initial_token")); |
| hosts.add(config.broadcastAddress()); |
| hostIds.add(config.hostId()); |
| } |
| |
| try |
| { |
| IPartitioner partitioner = FBUtilities.newPartitioner(partitionerName); |
| StorageService storageService = StorageService.instance; |
| List<Token> tokens = new ArrayList<>(); |
| for (String token : initialTokens) |
| tokens.add(partitioner.getTokenFactory().fromString(token)); |
| |
| for (int i = 0; i < tokens.size(); i++) |
| { |
| InetSocketAddress ep = hosts.get(i); |
| UUID hostId = hostIds.get(i); |
| Token token = tokens.get(i); |
| Gossiper.runInGossipStageBlocking(() -> { |
| Gossiper.instance.initializeNodeUnsafe(ep.getAddress(), hostId, 1); |
| Gossiper.instance.injectApplicationState(ep.getAddress(), |
| ApplicationState.TOKENS, |
| new VersionedValue.VersionedValueFactory(partitioner).tokens(Collections.singleton(token))); |
| storageService.onChange(ep.getAddress(), |
| ApplicationState.STATUS, |
| new VersionedValue.VersionedValueFactory(partitioner).normal(Collections.singleton(token))); |
| Gossiper.instance.realMarkAlive(ep.getAddress(), Gossiper.instance.getEndpointStateForEndpoint(ep.getAddress())); |
| }); |
| int messagingVersion = cluster.get(ep).isShutdown() |
| ? MessagingService.current_version |
| : Math.min(MessagingService.current_version, cluster.get(ep).getMessagingVersion()); |
| MessagingService.instance().setVersion(ep.getAddress(), messagingVersion); |
| } |
| |
| // check that all nodes are in token metadata |
| for (int i = 0; i < tokens.size(); ++i) |
| assert storageService.getTokenMetadata().isMember(hosts.get(i).getAddress()); |
| } |
| catch (Throwable e) // UnknownHostException |
| { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| public Future<Void> shutdown() |
| { |
| return shutdown(true); |
| } |
| |
| @Override |
| public Future<Void> shutdown(boolean graceful) |
| { |
| if (!graceful) |
| MessagingService.instance().shutdown(false); |
| |
| Future<?> future = async((ExecutorService executor) -> { |
| Throwable error = null; |
| |
| error = parallelRun(error, executor, |
| () -> StorageService.instance.setRpcReady(false), |
| CassandraDaemon.getInstanceForTesting()::destroyNativeTransport); |
| |
| if (config.has(GOSSIP) || config.has(NETWORK)) |
| { |
| StorageService.instance.shutdownServer(); |
| |
| error = parallelRun(error, executor, |
| () -> NanoTimeToCurrentTimeMillis.shutdown(MINUTES.toMillis(1L)) |
| ); |
| } |
| |
| error = parallelRun(error, executor, |
| () -> Gossiper.instance.stopShutdownAndWait(1L, MINUTES), |
| CompactionManager.instance::forceShutdown, |
| () -> BatchlogManager.instance.shutdownAndWait(1L, MINUTES), |
| HintsService.instance::shutdownBlocking, |
| () -> StreamCoordinator.shutdownAndWait(1L, MINUTES), |
| () -> SecondaryIndexManager.shutdownAndWait(1L, MINUTES), |
| () -> IndexSummaryManager.instance.shutdownAndWait(1L, MINUTES), |
| () -> ColumnFamilyStore.shutdownExecutorsAndWait(1L, MINUTES), |
| () -> PendingRangeCalculatorService.instance.shutdownExecutor(1L, MINUTES), |
| () -> BufferPool.shutdownLocalCleaner(1L, MINUTES), |
| () -> StorageService.instance.shutdownBGMonitorAndWait(1L, MINUTES), |
| () -> Ref.shutdownReferenceReaper(1L, MINUTES), |
| () -> Memtable.MEMORY_POOL.shutdownAndWait(1L, MINUTES), |
| () -> SSTableReader.shutdownBlocking(1L, MINUTES), |
| () -> DiagnosticSnapshotService.instance.shutdownAndWait(1L, MINUTES) |
| ); |
| error = parallelRun(error, executor, |
| () -> ScheduledExecutors.shutdownAndWait(1L, MINUTES), |
| MessagingService.instance()::shutdown |
| ); |
| error = parallelRun(error, executor, |
| () -> StageManager.shutdownAndWait(1L, MINUTES), |
| () -> SharedExecutorPool.SHARED.shutdownAndWait(1L, MINUTES) |
| ); |
| error = parallelRun(error, executor, |
| CommitLog.instance::shutdownBlocking |
| ); |
| |
| Throwables.maybeFail(error); |
| }).apply(isolatedExecutor); |
| |
| return CompletableFuture.runAsync(ThrowingRunnable.toRunnable(future::get), isolatedExecutor) |
| .thenRun(super::shutdown); |
| } |
| |
| public int liveMemberCount() |
| { |
| return sync(() -> { |
| if (!DatabaseDescriptor.isDaemonInitialized() || !Gossiper.instance.isEnabled()) |
| return 0; |
| return Gossiper.instance.getLiveMembers().size(); |
| }).call(); |
| } |
| |
| public NodeToolResult nodetoolResult(boolean withNotifications, String... commandAndArgs) |
| { |
| return sync(() -> { |
| DTestNodeTool nodetool = new DTestNodeTool(withNotifications); |
| int rc = nodetool.execute(commandAndArgs); |
| return new NodeToolResult(commandAndArgs, rc, new ArrayList<>(nodetool.notifications.notifications), nodetool.latestError); |
| }).call(); |
| } |
| |
| private static class DTestNodeTool extends NodeTool { |
| private final StorageServiceMBean storageProxy; |
| private final CollectingNotificationListener notifications = new CollectingNotificationListener(); |
| |
| private Throwable latestError; |
| |
| DTestNodeTool(boolean withNotifications) { |
| super(new InternalNodeProbeFactory(withNotifications)); |
| storageProxy = new InternalNodeProbe(withNotifications).getStorageService(); |
| storageProxy.addNotificationListener(notifications, null, null); |
| } |
| |
| public int execute(String... args) |
| { |
| try |
| { |
| return super.execute(args); |
| } |
| finally |
| { |
| try |
| { |
| storageProxy.removeNotificationListener(notifications, null, null); |
| } |
| catch (ListenerNotFoundException e) |
| { |
| // ignored |
| } |
| } |
| } |
| |
| protected void badUse(Exception e) |
| { |
| super.badUse(e); |
| latestError = e; |
| } |
| |
| protected void err(Throwable e) |
| { |
| super.err(e); |
| latestError = e; |
| } |
| } |
| |
| private static final class CollectingNotificationListener implements NotificationListener |
| { |
| private final List<Notification> notifications = new CopyOnWriteArrayList<>(); |
| |
| public void handleNotification(Notification notification, Object handback) |
| { |
| notifications.add(notification); |
| } |
| } |
| |
| public void uncaughtException(Thread thread, Throwable throwable) |
| { |
| System.out.println(String.format("Exception %s occurred on thread %s", throwable.getMessage(), thread.getName())); |
| throwable.printStackTrace(); |
| } |
| |
| public long killAttempts() |
| { |
| return callOnInstance(InstanceKiller::getKillAttempts); |
| } |
| |
| 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) |
| { |
| List<Future<Throwable>> results = new ArrayList<>(); |
| for (ThrowingRunnable runnable : runnables) |
| { |
| results.add(runOn.submit(() -> { |
| try |
| { |
| runnable.run(); |
| return null; |
| } |
| catch (Throwable t) |
| { |
| return t; |
| } |
| })); |
| } |
| for (Future<Throwable> future : results) |
| { |
| try |
| { |
| Throwable t = future.get(); |
| if (t != null) |
| throw t; |
| } |
| catch (Throwable t) |
| { |
| accumulate = Throwables.merge(accumulate, t); |
| } |
| } |
| return accumulate; |
| } |
| } |
