fix regressions

patch by jbellis reviewed by Eric Evans for CASSANDRA-477

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@823617 13f79535-47bb-0310-9956-ffa450edef68

src/java/org/apache/cassandra/dht/BootStrapper.java

@@ -37,6 +37,7 @@ package org.apache.cassandra.dht;
  import org.apache.commons.lang.ArrayUtils;
 
  import org.apache.cassandra.locator.TokenMetadata;
+ import org.apache.cassandra.locator.AbstractReplicationStrategy;
  import org.apache.cassandra.net.*;
  import org.apache.cassandra.net.io.StreamContextManager;
  import org.apache.cassandra.net.io.IStreamComplete;
@@ -208,6 +209,7 @@ public class BootStrapper implements Runnable
 
             if (!maxEndpoint.equals(StorageService.getLocalStorageEndPoint()))
             {
+                StorageService.instance().retrofitPorts(Arrays.asList(maxEndpoint));
                 Token<?> t = getBootstrapTokenFrom(maxEndpoint);
                 logger_.info("Setting token to " + t + " to assume load from " + maxEndpoint.getHost());
                 ss.updateToken(t);

src/java/org/apache/cassandra/gms/Gossiper.java

@@ -95,7 +95,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
     final static String GOSSIP_DIGEST_ACK_VERB = "GAV";
     /* GA2V - abbreviation for GOSSIP-DIGEST-ACK2-VERB */
     final static String GOSSIP_DIGEST_ACK2_VERB = "GA2V";
-    final static int intervalInMillis_ = 1000;
+    public final static int intervalInMillis_ = 1000;
     private static Logger logger_ = Logger.getLogger(Gossiper.class);
     static Gossiper gossiper_;
 
@@ -522,6 +522,8 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC
                     {
                         reqdEndPointState = new EndPointState(epState.getHeartBeatState());
                     }
+                    if (logger_.isTraceEnabled())
+                        logger_.trace("Adding state " + key + ": " + appState.getState());
                     reqdEndPointState.addApplicationState(key, appState);
                 }
             }

src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java

@@ -73,7 +73,7 @@ public abstract class AbstractReplicationStrategy
      * This method changes the ports of the endpoints from
      * the control port to the storage ports.
     */
-    protected void retrofitPorts(List<EndPoint> eps)
+    public void retrofitPorts(List<EndPoint> eps)
     {
         for ( EndPoint ep : eps )
         {

src/java/org/apache/cassandra/service/StorageLoadBalancer.java

@@ -351,8 +351,9 @@ public final class StorageLoadBalancer implements IEndPointStateChangeSubscriber
 
     public void startBroadcasting()
     {
-        /* starts a load timer thread */
+        // send the first broadcast "right away" (i.e., in 2 gossip heartbeats, when we should have someone to talk to);
-        loadTimer_.schedule(new LoadDisseminator(), BROADCAST_INTERVAL, BROADCAST_INTERVAL);
+        // after that send every BROADCAST_INTERVAL.
+        loadTimer_.schedule(new LoadDisseminator(), 2 * Gossiper.intervalInMillis_, BROADCAST_INTERVAL);
     }
 
     /** wait for node information to be available.  if the rest of the cluster just came up,

src/java/org/apache/cassandra/service/StorageService.java

@@ -226,12 +226,12 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
         MessagingService.instance().registerVerbHandlers(mbrshipCleanerVerbHandler_, new MembershipCleanerVerbHandler() );
         MessagingService.instance().registerVerbHandlers(rangeVerbHandler_, new RangeVerbHandler());
         // see BootStrapper for a summary of how the bootstrap verbs interact
+        MessagingService.instance().registerVerbHandlers(bootstrapTokenVerbHandler_, new BootStrapper.BootstrapTokenVerbHandler());
         MessagingService.instance().registerVerbHandlers(bootstrapMetadataVerbHandler_, new BootstrapMetadataVerbHandler() );
         MessagingService.instance().registerVerbHandlers(bootStrapInitiateVerbHandler_, new BootStrapper.BootStrapInitiateVerbHandler());
         MessagingService.instance().registerVerbHandlers(bootStrapInitiateDoneVerbHandler_, new BootStrapper.BootstrapInitiateDoneVerbHandler());
         MessagingService.instance().registerVerbHandlers(bootStrapTerminateVerbHandler_, new BootStrapper.BootstrapTerminateVerbHandler());
-        MessagingService.instance().registerVerbHandlers(bootstrapTokenVerbHandler_, new BootStrapper.BootstrapTokenVerbHandler());
+
-        
         StageManager.registerStage(StorageService.mutationStage_,
                                    new MultiThreadedStage(StorageService.mutationStage_, DatabaseDescriptor.getConcurrentWriters()));
         StageManager.registerStage(StorageService.readStage_,
@@ -266,15 +266,24 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
 
         StorageLoadBalancer.instance().startBroadcasting();
 
-        if (isBootstrapMode)
+        // have to start the gossip service before we can see any info on other nodes.  this is necessary
-        {
+        // for bootstrap to get the load info it needs.
-            BootStrapper.startBootstrap();
+        // (we won't be part of the storage ring though until we add a nodeId to our state, below.)
-        }
-
         Gossiper.instance().register(this);
         Gossiper.instance().start(udpAddr_, storageMetadata_.getGeneration());
-        /* Make sure this token gets gossiped around. */
+
-        tokenMetadata_.update(storageMetadata_.getToken(), StorageService.tcpAddr_, isBootstrapMode);
+        if (isBootstrapMode)
+        {
+            BootStrapper.startBootstrap(); // handles token update
+        }
+        else
+        {
+            tokenMetadata_.update(storageMetadata_.getToken(), StorageService.tcpAddr_, isBootstrapMode);
+        }
+
+        // Gossip my token.
+        // note that before we do this we've (a) finalized what the token is actually going to be, and
+        // (b) added a bootstrap state (done by startBootstrap)
         ApplicationState state = new ApplicationState(StorageService.getPartitioner().getTokenFactory().toString(storageMetadata_.getToken()));
         Gossiper.instance().addApplicationState(StorageService.nodeId_, state);
     }
@@ -935,6 +944,11 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
         return nodePicker_.getHintedStorageEndPoints(partitioner_.getToken(key));
     }
 
+    public void retrofitPorts(List<EndPoint> eps)
+    {
+        nodePicker_.retrofitPorts(eps);
+    }
+
     /**
      * This function finds the most suitable endpoint given a key.
      * It checks for locality and alive test.