Fix #536. New nodes should add old shards without panicing

CHANGELOG.md

@@ -4,8 +4,9 @@
 
 - [Issue #516](https://github.com/influxdb/influxdb/issues/516). Close WAL log/index files when they aren't being used
 - [Issue #532](https://github.com/influxdb/influxdb/issues/532). Don't log graphite connection EOF as an error
-- [Issue #535](https://github.com/influxdb/influxdb/pull/535). WAL Replay hangs if response isn't received
-- [Issue #538](https://github.com/influxdb/influxdb/pull/538). Don't panic if the same series existed twice in the request with different columns
+- [Issue #535](https://github.com/influxdb/influxdb/issues/535). WAL Replay hangs if response isn't received
+- [Issue #538](https://github.com/influxdb/influxdb/issues/538). Don't panic if the same series existed twice in the request with different columns
+- [Issue #536](https://github.com/influxdb/influxdb/issues/536). Joining the cluster after shards are creating shouldn't cause new nodes to panic
 
 ## v0.6.2 [2014-05-09]
 

src/cluster/cluster_configuration.go

@@ -890,7 +890,10 @@ func (self *ClusterConfiguration) AddShards(shards []*NewShardData) ([]*ShardDat
 		shard := NewShard(id, newShard.StartTime, newShard.EndTime, shardType, durationIsSplit, self.wal)
 		servers := make([]*ClusterServer, 0)
 		for _, serverId := range newShard.ServerIds {
-			if serverId == self.LocalServer.Id {
+			// if a shard is created before the local server then the local
+			// server can't be one of the servers the shard belongs to,
+			// since the shard was created before the server existed
+			if self.LocalServer != nil && serverId == self.LocalServer.Id {
 				err := shard.SetLocalStore(self.shardStore, self.LocalServer.Id)
 				if err != nil {
 					log.Error("AddShards: error setting local store: ", err)

src/integration/late_join_test.go

@@ -0,0 +1,44 @@
+package integration
+
+import (
+	. "integration/helpers"
+	"os"
+	. "launchpad.net/gocheck"
+)
+
+type LateJoinSuite struct {
+	serverProcesses []*Server
+}
+
+var _ = Suite(&LateJoinSuite{})
+
+func (self *LateJoinSuite) SetUpSuite(c *C) {
+}
+
+func (self *LateJoinSuite) TearDownSuite(c *C) {
+	for _, s := range self.serverProcesses {
+		s.Stop()
+	}
+}
+
+func (self *LateJoinSuite) TestReplayingWAL(c *C) {
+	err := os.RemoveAll("/tmp/influxdb/test")
+	c.Assert(err, IsNil)
+	s1 := NewServer("src/integration/test_rf_1.toml", c)
+	defer s1.Stop()
+
+	client := s1.GetClient("", c)
+	c.Assert(client.CreateDatabase("test"), IsNil)
+	series := CreatePoints("test_replication_factor_1", 1, 1)
+	c.Assert(client.WriteSeries(series), IsNil)
+
+	s2 := NewServer("src/integration/test_rf_2.toml", c)
+	defer s2.Stop()
+
+	s2.WaitForServerToStart()
+	s1.WaitForServerToSync()
+	s2.WaitForServerToSync()
+
+	client = s2.GetClient("", c)
+	c.Assert(client.Ping(), IsNil)
+}