diff --git a/bin/tokenupdater b/bin/tokenupdater deleted file mode 100755 index 12118e9791..0000000000 --- a/bin/tokenupdater +++ /dev/null @@ -1,49 +0,0 @@ -#!/bin/sh -# 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. - - -if [ "x$CASSANDRA_INCLUDE" = "x" ]; then - for include in /usr/share/cassandra/cassandra.in.sh \ - /usr/local/share/cassandra/cassandra.in.sh \ - /opt/cassandra/cassandra.in.sh \ - `dirname $0`/cassandra.in.sh; do - if [ -r $include ]; then - . $include - break - fi - done -elif [ -r $CASSANDRA_INCLUDE ]; then - . $CASSANDRA_INCLUDE -fi - -if [ -z $CASSANDRA_CONF -o -z $CLASSPATH ]; then - echo "You must set the CASSANDRA_CONF and CLASSPATH vars" >&2 - exit 1 -fi - -# Special-case path variables. -case "`uname`" in - CYGWIN*) - CLASSPATH=`cygpath -p -w "$CLASSPATH"` - CASSANDRA_CONF=`cygpath -p -w "$CASSANDRA_CONF"` - ;; -esac - -java -cp $CLASSPATH -Dstorage-config=$CASSANDRA_CONF \ - org.apache.cassandra.tools.TokenUpdater $@ - -# vi:ai sw=4 ts=4 tw=0 et diff --git a/src/java/org/apache/cassandra/db/SystemTable.java b/src/java/org/apache/cassandra/db/SystemTable.java index 61dbbdc6b7..c03535f37c 100644 --- a/src/java/org/apache/cassandra/db/SystemTable.java +++ b/src/java/org/apache/cassandra/db/SystemTable.java @@ -127,7 +127,7 @@ public class SystemTable if (initialToken == null) token = p.getRandomToken(); else - token = p.getToken(initialToken); + token = p.getTokenFactory().fromString(initialToken); logger.info("Saved Token not found. Using " + token); // seconds-since-epoch isn't a foolproof new generation diff --git a/src/java/org/apache/cassandra/dht/IPartitioner.java b/src/java/org/apache/cassandra/dht/IPartitioner.java index 27469d5a8f..6cb31350b2 100644 --- a/src/java/org/apache/cassandra/dht/IPartitioner.java +++ b/src/java/org/apache/cassandra/dht/IPartitioner.java @@ -72,6 +72,8 @@ public interface IPartitioner /** * @return a Token that can be used to route a given key + * (This is NOT a method to create a Token from its string representation; + * for that, use TokenFactory.fromString.) */ public T getToken(String key); diff --git a/src/java/org/apache/cassandra/io/SSTable.java b/src/java/org/apache/cassandra/io/SSTable.java index 4f60c3773b..43d41cac03 100644 --- a/src/java/org/apache/cassandra/io/SSTable.java +++ b/src/java/org/apache/cassandra/io/SSTable.java @@ -50,6 +50,7 @@ public abstract class SSTable { private static final Logger logger = Logger.getLogger(SSTable.class); + public static final int FILES_ON_DISK = 3; // data, index, and bloom filter protected String path; protected IPartitioner partitioner; diff --git a/src/java/org/apache/cassandra/io/Streaming.java b/src/java/org/apache/cassandra/io/Streaming.java index 31f2a04b0f..c3b00f04fb 100644 --- a/src/java/org/apache/cassandra/io/Streaming.java +++ b/src/java/org/apache/cassandra/io/Streaming.java @@ -72,7 +72,7 @@ public class Streaming if (sstables.isEmpty()) return; - StreamContextManager.StreamContext[] streamContexts = new StreamContextManager.StreamContext[sstables.size()]; + StreamContextManager.StreamContext[] streamContexts = new StreamContextManager.StreamContext[SSTable.FILES_ON_DISK * sstables.size()]; int i = 0; for (SSTableReader sstable : sstables) { @@ -130,7 +130,6 @@ public class Streaming StreamContextManager.StreamStatus streamStatus = new StreamContextManager.StreamStatus(streamContext.getTargetFile(), streamContext.getExpectedBytes() ); String file = getNewFileNameFromOldContextAndNames(fileNames, streamContext); - //String file = DatabaseDescriptor.getDataFileLocationForTable(streamContext.getTable()) + File.separator + newFileName + "-Data.db"; if (logger.isDebugEnabled()) logger.debug("Received Data from : " + message.getFrom() + " " + streamContext.getTargetFile() + " " + file); streamContext.setTargetFile(file); diff --git a/src/java/org/apache/cassandra/locator/TokenMetadata.java b/src/java/org/apache/cassandra/locator/TokenMetadata.java index 5fd7eca943..164d80fde6 100644 --- a/src/java/org/apache/cassandra/locator/TokenMetadata.java +++ b/src/java/org/apache/cassandra/locator/TokenMetadata.java @@ -88,6 +88,7 @@ public class TokenMetadata lock.writeLock().lock(); try { + tokenToEndPointMap.inverse().remove(endpoint); if (!endpoint.equals(tokenToEndPointMap.put(token, endpoint))) { sortedTokens = sortTokens(); diff --git a/src/java/org/apache/cassandra/net/TcpConnection.java b/src/java/org/apache/cassandra/net/TcpConnection.java index 62a73a22fd..ae6601bc23 100644 --- a/src/java/org/apache/cassandra/net/TcpConnection.java +++ b/src/java/org/apache/cassandra/net/TcpConnection.java @@ -71,7 +71,6 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable private TcpConnection(InetAddress from, InetAddress to, TcpConnectionManager pool, boolean streaming) throws IOException { - logger_.debug("creating connection from " + from + " to " + to); socketChannel_ = SocketChannel.open(); socketChannel_.socket().bind(new InetSocketAddress(from, 0)); socketChannel_.configureBlocking(false); @@ -291,18 +290,20 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable void closeSocket() { - logger_.warn("Closing down connection " + socketChannel_ + " with " + pendingWrites_.size() + " writes remaining."); + if (pendingWrites_.size() > 0) + logger_.error("Closing down connection " + socketChannel_ + " with " + pendingWrites_.size() + " writes remaining."); cancel(key_); pendingWrites_.clear(); } void errorClose() { - logger_.warn("Closing down connection " + socketChannel_); + logger_.info("Closing errored connection " + socketChannel_); pendingWrites_.clear(); cancel(key_); pendingWrites_.clear(); - pool_.destroy(this); + if (pool_ != null) + pool_.destroy(this); } private void cancel(SelectionKey key) diff --git a/src/java/org/apache/cassandra/net/io/StreamContextManager.java b/src/java/org/apache/cassandra/net/io/StreamContextManager.java index 28bb4f315f..4723afcc04 100644 --- a/src/java/org/apache/cassandra/net/io/StreamContextManager.java +++ b/src/java/org/apache/cassandra/net/io/StreamContextManager.java @@ -261,7 +261,7 @@ public class StreamContextManager { List context = ctxBag_.get(key); if ( context == null ) - throw new IllegalStateException("Streaming context has not been set."); + throw new IllegalStateException("Streaming context has not been set for " + key); StreamContext streamContext = context.remove(0); if ( context.isEmpty() ) ctxBag_.remove(key); @@ -272,7 +272,7 @@ public class StreamContextManager { List status = streamStatusBag_.get(key); if ( status == null ) - throw new IllegalStateException("Streaming status has not been set."); + throw new IllegalStateException("Streaming status has not been set for " + key); StreamStatus streamStatus = status.remove(0); if ( status.isEmpty() ) streamStatusBag_.remove(key); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 36c6ebc4fa..6a51332f0a 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -19,6 +19,7 @@ package org.apache.cassandra.service; import java.io.IOException; +import java.io.IOError; import java.lang.management.ManagementFactory; import java.lang.reflect.Constructor; import java.util.*; @@ -71,7 +72,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto /* All verb handler identifiers */ public final static String mutationVerbHandler_ = "ROW-MUTATION-VERB-HANDLER"; - public final static String tokenVerbHandler_ = "TOKEN-VERB-HANDLER"; public final static String binaryVerbHandler_ = "BINARY-VERB-HANDLER"; public final static String readRepairVerbHandler_ = "READ-REPAIR-VERB-HANDLER"; public final static String readVerbHandler_ = "ROW-READ-VERB-HANDLER"; @@ -165,7 +165,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto { bootstrapSet.remove(s); if (logger_.isDebugEnabled()) - logger_.debug("Removed " + s + " as a bootstrap source"); + logger_.debug("Removed " + s + " as a bootstrap source; remaining is [" + StringUtils.join(bootstrapSet, ", ") + "]"); if (bootstrapSet.isEmpty()) { @@ -222,7 +222,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto endPointSnitch_ = DatabaseDescriptor.getEndPointSnitch(); /* register the verb handlers */ - MessagingService.instance().registerVerbHandlers(tokenVerbHandler_, new TokenUpdateVerbHandler()); MessagingService.instance().registerVerbHandlers(binaryVerbHandler_, new BinaryVerbHandler()); MessagingService.instance().registerVerbHandlers(mutationVerbHandler_, new RowMutationVerbHandler()); MessagingService.instance().registerVerbHandlers(readRepairVerbHandler_, new ReadRepairVerbHandler()); @@ -258,8 +257,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto public void start() throws IOException { storageMetadata_ = SystemTable.initMetadata(); - isBootstrapMode = DatabaseDescriptor.isAutoBootstrap() - && !(DatabaseDescriptor.getSeeds().contains(FBUtilities.getLocalAddress()) || SystemTable.isBootstrapped()); /* Listen for application messages */ MessagingService.instance().listen(FBUtilities.getLocalAddress()); @@ -277,15 +274,14 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto Gossiper.instance().register(this); Gossiper.instance().start(FBUtilities.getLocalAddress(), storageMetadata_.getGeneration()); - if (isBootstrapMode) + if (DatabaseDescriptor.isAutoBootstrap() + && !(DatabaseDescriptor.getSeeds().contains(FBUtilities.getLocalAddress()) || SystemTable.isBootstrapped())) { logger_.info("Starting in bootstrap mode (first, sleeping to get load information)"); StorageLoadBalancer.instance().waitForLoadInfo(); logger_.info("... got load info"); Token token = BootStrapper.getBootstrapToken(tokenMetadata_, StorageLoadBalancer.instance().getLoadInfo()); - SystemTable.updateToken(token); // DON'T use setToken, that makes us part of the ring locally which is incorrect until we are done bootstrapping - Gossiper.instance().addApplicationState(StorageService.STATE_BOOTSTRAPPING, new ApplicationState(partitioner_.getTokenFactory().toString(getLocalToken()))); - new BootStrapper(replicationStrategy_, FBUtilities.getLocalAddress(), getLocalToken(), tokenMetadata_).startBootstrap(); // handles token update + startBootstrap(token); // don't finish startup (enabling thrift) until after bootstrap is done while (isBootstrapMode) { @@ -303,13 +299,21 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto { SystemTable.setBootstrapped(true); Token token = storageMetadata_.getToken(); - setToken(token); + tokenMetadata_.update(token, FBUtilities.getLocalAddress()); Gossiper.instance().addApplicationState(StorageService.STATE_NORMAL, new ApplicationState(partitioner_.getTokenFactory().toString(token))); } assert tokenMetadata_.sortedTokens().size() > 0; } + private void startBootstrap(Token token) throws IOException + { + isBootstrapMode = true; + SystemTable.updateToken(token); // DON'T use setToken, that makes us part of the ring locally which is incorrect until we are done bootstrapping + Gossiper.instance().addApplicationState(StorageService.STATE_BOOTSTRAPPING, new ApplicationState(partitioner_.getTokenFactory().toString(token))); + new BootStrapper(replicationStrategy_, FBUtilities.getLocalAddress(), token, tokenMetadata_).startBootstrap(); // handles token update + } + public boolean isBootstrapMode() { return isBootstrapMode; @@ -935,7 +939,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto return tokens; } - public void decommission() + public void decommission() throws InterruptedException { if (!tokenMetadata_.isMember(FBUtilities.getLocalAddress())) throw new UnsupportedOperationException("local node is not a member of the token ring yet"); @@ -947,15 +951,22 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto logger_.info("DECOMMISSIONING"); Gossiper.instance().addApplicationState(STATE_LEAVING, new ApplicationState(getLocalToken().toString())); logger_.info("decommission sleeping " + Streaming.RING_DELAY); - try - { - Thread.sleep(Streaming.RING_DELAY); - } - catch (InterruptedException e) - { - throw new AssertionError(e); - } + Thread.sleep(Streaming.RING_DELAY); + Runnable finishLeaving = new Runnable() + { + public void run() + { + Gossiper.instance().stop(); + logger_.info("DECOMMISSION FINISHED."); + // let op be responsible for killing the process + } + }; + unbootstrap(finishLeaving); + } + + private void unbootstrap(final Runnable onFinish) + { Multimap rangesMM = getChangedRangesForLeaving(FBUtilities.getLocalAddress()); if (logger_.isDebugEnabled()) logger_.debug("Ranges needing transfer are [" + StringUtils.join(rangesMM.keySet(), ",") + "]"); @@ -970,7 +981,25 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto { pending.remove(entry); if (pending.isEmpty()) - finishLeaving(); + { + SystemTable.setBootstrapped(false); + tokenMetadata_.removeEndpoint(FBUtilities.getLocalAddress()); + replicationStrategy_.removeObsoletePendingRanges(); + + if (logger_.isDebugEnabled()) + logger_.debug(""); + Gossiper.instance().addApplicationState(STATE_LEFT, new ApplicationState(getLocalToken().toString())); + try + { + Thread.sleep(2 * Gossiper.intervalInMillis_); + } + catch (InterruptedException e) + { + throw new AssertionError(e); + } + + onFinish.run(); + } } }; StageManager.getStage(streamStage_).execute(new Runnable() @@ -984,6 +1013,35 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto } } + public void move(String newToken) throws InterruptedException + { + if (tokenMetadata_.getPendingRanges(FBUtilities.getLocalAddress()).size() > 0) + throw new UnsupportedOperationException("data is currently moving to this node; unable to leave the ring"); + + final Token token = partitioner_.getTokenFactory().fromString(newToken); // make sure it's valid + logger_.info("moving to " + token); + Gossiper.instance().addApplicationState(STATE_LEAVING, new ApplicationState(getLocalToken().toString())); + logger_.info("move sleeping " + Streaming.RING_DELAY); + Thread.sleep(Streaming.RING_DELAY); + + Runnable finishMoving = new Runnable() + { + public void run() + { + try + { + logger_.info("re-bootstrapping to new token " + token); + startBootstrap(token); + } + catch (IOException e) + { + throw new IOError(e); + } + } + }; + unbootstrap(finishMoving); + } + public QuorumResponseHandler getResponseHandler(IResponseResolver responseResolver, int blockFor, int consistency_level) { return replicationStrategy_.getResponseHandler(responseResolver, blockFor, consistency_level); @@ -994,20 +1052,4 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto return replicationStrategy_; } - public void finishLeaving() - { - SystemTable.setBootstrapped(false); - Gossiper.instance().addApplicationState(STATE_LEFT, new ApplicationState(getLocalToken().toString())); - try - { - Thread.sleep(2 * Gossiper.intervalInMillis_); - } - catch (InterruptedException e) - { - throw new AssertionError(e); - } - Gossiper.instance().stop(); - logger_.info("DECOMMISSION FINISHED."); - // let op be responsible for killing the process - } } diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index 01c9657def..dc458fd300 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -23,6 +23,7 @@ import java.net.UnknownHostException; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.FutureTask; import org.apache.cassandra.dht.Range; import java.net.InetAddress; @@ -116,7 +117,13 @@ public interface StorageServiceMBean /** * transfer this node's data to other machines and remove it from service. */ - public void decommission(); + public void decommission() throws InterruptedException; + + /** + * @param newToken token to move this node to. + * This node will unload its data onto its neighbors, and bootstrap to the new token. + */ + public void move(String newToken) throws InterruptedException; /** set the logging level at runtime */ public void setLog4jLevel(String classQualifier, String level); diff --git a/src/java/org/apache/cassandra/service/TokenUpdateVerbHandler.java b/src/java/org/apache/cassandra/service/TokenUpdateVerbHandler.java deleted file mode 100644 index 7a4af4b438..0000000000 --- a/src/java/org/apache/cassandra/service/TokenUpdateVerbHandler.java +++ /dev/null @@ -1,56 +0,0 @@ -/** - * 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.service; - -import java.io.IOException; - -import org.apache.log4j.Logger; - -import org.apache.cassandra.dht.Token; -import org.apache.cassandra.io.DataInputBuffer; -import org.apache.cassandra.net.IVerbHandler; -import org.apache.cassandra.net.Message; -import org.apache.cassandra.gms.Gossiper; -import org.apache.cassandra.gms.ApplicationState; - -public class TokenUpdateVerbHandler implements IVerbHandler -{ - private static Logger logger_ = Logger.getLogger(TokenUpdateVerbHandler.class); - - public void doVerb(Message message) - { - if (StorageService.instance().isBootstrapMode()) - throw new UnsupportedOperationException("Cannot set token during bootstrap"); - - byte[] body = message.getMessageBody(); - DataInputBuffer bufIn = new DataInputBuffer(); - bufIn.reset(body, body.length); - try - { - Token token = Token.serializer().deserialize(bufIn); - StorageService.instance().setToken(token); - Gossiper.instance().addApplicationState(StorageService.STATE_NORMAL, new ApplicationState(StorageService.getPartitioner().getTokenFactory().toString(token))); - } - catch (IOException ex) - { - throw new RuntimeException(ex); - } - } - -} diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index 28904d1fcb..a1a109c373 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -380,11 +380,16 @@ public class NodeProbe ssProxy.clearSnapshot(); } - public void decommission() + public void decommission() throws InterruptedException { ssProxy.decommission(); } + public void move(String newToken) throws InterruptedException + { + ssProxy.move(newToken); + } + /** * Print out the size of the queues in the thread pools * @@ -480,7 +485,7 @@ public class NodeProbe HelpFormatter hf = new HelpFormatter(); String header = String.format( "%nAvailable commands: ring, info, cleanup, compact, cfstats, snapshot [name], clearsnapshot, " + - "tpstats, flush, decommission, " + + "tpstats, flush, decommission, move, " + " getcompactionthreshold, setcompactionthreshold [minthreshold] ([maxthreshold])"); String usage = String.format("java %s -host %n", NodeProbe.class.getName()); hf.printHelp(usage, "", options, header); @@ -489,7 +494,7 @@ public class NodeProbe /** * @param args */ - public static void main(String[] args) throws IOException + public static void main(String[] args) throws IOException, InterruptedException { NodeProbe probe = null; try @@ -543,6 +548,14 @@ public class NodeProbe { probe.decommission(); } + else if (cmdName.equals("move")) + { + if (arguments.length <= 1) + { + System.err.println("missing token argument"); + } + probe.move(arguments[1]); + } else if (cmdName.equals("snapshot")) { String snapshotName = ""; diff --git a/src/java/org/apache/cassandra/tools/TokenUpdater.java b/src/java/org/apache/cassandra/tools/TokenUpdater.java deleted file mode 100644 index 5a7c513c37..0000000000 --- a/src/java/org/apache/cassandra/tools/TokenUpdater.java +++ /dev/null @@ -1,86 +0,0 @@ -/** - * 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.tools; - -import java.io.ByteArrayOutputStream; -import java.io.DataOutputStream; - -import org.apache.cassandra.dht.IPartitioner; -import org.apache.cassandra.dht.Token; -import java.net.InetAddress; -import java.net.InetSocketAddress; - -import org.apache.cassandra.net.Message; -import org.apache.cassandra.net.MessagingService; -import org.apache.cassandra.net.SelectorManager; -import org.apache.cassandra.service.StorageService; -import org.apache.cassandra.utils.FBUtilities; -import org.apache.cassandra.utils.FileUtils; - -public class TokenUpdater -{ - private static final int port_ = 7000; - private static final long waitTime_ = 10000; - - public static void main(String[] args) throws Throwable - { - if (args.length < 2) - { - System.out.println("Usage : java org.apache.cassandra.tools.TokenUpdater "); - System.exit(1); - } - - Thread selectorThread = SelectorManager.getSelectorManager(); - selectorThread.setDaemon(true); - selectorThread.start(); - - String ipPort = args[0]; - IPartitioner p = StorageService.getPartitioner(); - Token token = p.getTokenFactory().fromString(args[1]); - System.out.println("Partitioner is " + p.getClass() + ", token is: " + token); - System.out.println(p.getTokenFactory().getClass()); - - String[] ipPortPair = ipPort.split(":"); - int port = 7000; - if (ipPortPair.length > 1) - { - port = Integer.valueOf(ipPortPair[1]); - } - - InetSocketAddress target = new InetSocketAddress(ipPortPair[0], port); - - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - DataOutputStream dos = new DataOutputStream(bos); - Token.serializer().serialize(token, dos); - - /* Construct the token update message to be sent */ - Message tokenUpdateMessage = new Message(target.getAddress(), - "", - StorageService.tokenVerbHandler_, - bos.toByteArray()); - - System.out.println("Sending a token update message to " + target); - MessagingService.instance().sendOneWay(tokenUpdateMessage, target.getAddress()); - Thread.sleep(TokenUpdater.waitTime_); - System.out.println("Done sending the update message"); - - MessagingService.shutdown(); - FileUtils.shutdown(); - } -}