Repository: gocql/gocql (go)

Issue:
panic: first replica is not the primary replica for the token
Please answer these questions before submitting your issue. Thanks!

### What version of Cassandra are you using?

[cqlsh 5.0.1 | Cassandra 3.0.14.32 | CQL spec 3.4.0 | Native protocol v4]

keyspace replica: replication = {'class': 'NetworkTopologyStrategy', 'dca1': '2'}

### What version of Gocql are you using?

revision = "16cf9ea1b3e28090d416d36528f304e4553e6b56"

### What did you do?

Change DCAwareRoundRobinPolicy to TokenAwareHostPolicy(DCAwareRoundRobinPolicy)

### What did you expect to see?
Correctly initialized session for read/write

### What did you see instead?

This is different from the similar issue I found from this repo, there is no node add/remove, when we deploy the change and try to create session, the exception was thrown.

panic: first replica is not the primary replica for the token

goroutine 971 [running]:
code.uber.internal/infra/jaeger/vendor/github.com/gocql/gocql.(*networkTopology).replicaMap(0xc01c5445b8, 0xc010df5b00, 0x83, 0x83, 0xc030916000, 0x8300, 0x9d55, 0xc000bf2000)
	code.uber.internal/infra/jaeger/vendor/github.com/gocql/gocql/topology.go:203 +0x1366
code.uber.internal/infra/jaeger/vendor/github.com/gocql/gocql.(*tokenAwareHostPolicy).updateReplicas(0xc00257e000, 0xc002862310, 0xc0015a2320, 0xe)
	code.uber.internal/infra/jaeger/vendor/github.com/gocql/gocql/policies.go:468 +0x2b9
code.uber.internal/infra/jaeger/vendor/github.com/gocql/gocql.(*tokenAwareHostPolicy).KeyspaceChanged(0xc00257e000, 0xc0015a2320, 0xe, 0x0, 0x0)
	code.uber.internal/infra/jaeger/vendor/github.com/gocql/gocql/policies.go:452 +0x9a
code.uber.internal/infra/jaeger/vendor/github.com/gocql/gocql.(*Session).init(0xc008536e00, 0xc00257e070, 0x0)
	code.uber.internal/infra/jaeger/vendor/github.com/gocql/gocql/session.go:256 +0x3cb
code.uber.internal/infra/jaeger/vendor/github.com/gocql/gocql.NewSession(0xc00274f000, 0x41, 0x80, 0x167581a, 0x5, 0x4, 0x2540be400, 0x0, 0x6b4c, 0xc0015a2320, ...)
	code.uber.internal/infra/jaeger/vendor/github.com/gocql/gocql/session.go:161 +0x824
code.uber.internal/infra/jaeger/vendor/github.com/gocql/gocql.(*ClusterConfig).CreateSession(...)
	code.uber.internal/infra/jaeger/vendor/github.com/gocql/gocql/cluster.go:185
code.uber.internal/infra/jaeger/cmd/ingester/app/cassandra.NewSession.func1(0xc0000d3f80, 0x16b4879, 0x2e, 0x0)
---

If you are having connectivity related issues please share the following additional information

### Describe your Cassandra cluster
please provide the following information

- output of `nodetool status`
- output of `SELECT peer, rpc_address FROM system.peers`
- rebuild your application with the `gocql_debug` tag and post the output


Pull request:
allow tokenAware policy to handle keyspace with skipped dcs
Keyspace could define replication that doesn't cover all datacenters.
More over: such case certainly happens when new data is added to cluster.

This commit fixes handling of this situation in networkTopology replicaMap
by skipping tokens from excess datacenters.

fixes #1349

Patch:
diff --git a/AUTHORS b/AUTHORS
index 5747e19..4bf368d 100644
--- a/AUTHORS
+++ b/AUTHORS
@@ -113,3 +113,4 @@ Thomas Meson <zllak@hycik.org>
 Martin Sucha <martin.sucha@kiwi.com>; <git@mm.ms47.eu>
 Pavel Buchinchik <p.buchinchik@gmail.com>
 Rintaro Okamura <rintaro.okamura@gmail.com>
+Yura Sokolov <y.sokolov@joom.com>; <funny.falcon@gmail.com>
diff --git a/topology.go b/topology.go
index 5ff7fd2..3f8a75e 100644
--- a/topology.go
+++ b/topology.go
@@ -171,7 +171,7 @@ func (n *networkTopology) replicaMap(tokenRing *tokenRing) tokenRingReplicas {
 	}
 
 	tokens := tokenRing.tokens
-	replicaRing := make(tokenRingReplicas, len(tokens))
+	replicaRing := make(tokenRingReplicas, 0, len(tokens))
 
 	var totalRF int
 	for _, rf := range n.dcs {
@@ -179,6 +179,11 @@ func (n *networkTopology) replicaMap(tokenRing *tokenRing) tokenRingReplicas {
 	}
 
 	for i, th := range tokenRing.tokens {
+		if rf := n.dcs[th.host.DataCenter()]; rf == 0 {
+			// skip this token since no replica in this datacenter.
+			continue
+		}
+
 		for k, v := range skipped {
 			skipped[k] = v[:0]
 		}
@@ -202,9 +207,9 @@ func (n *networkTopology) replicaMap(tokenRing *tokenRing) tokenRingReplicas {
 			dc := h.DataCenter()
 			rack := h.Rack()
 
-			rf, ok := n.dcs[dc]
-			if !ok {
-				// skip this DC, dont know about it
+			rf := n.dcs[dc]
+			if rf == 0 {
+				// skip this DC, dont know about it or replication factor is zero
 				continue
 			} else if replicasInDC[dc] >= rf {
 				if replicasInDC[dc] > rf {
@@ -261,10 +266,10 @@ func (n *networkTopology) replicaMap(tokenRing *tokenRing) tokenRingReplicas {
 			panic(fmt.Sprintf("first replica is not the primary replica for the token: expected %v got %v", replicas[0].ConnectAddress(), th.host.ConnectAddress()))
 		}
 
-		replicaRing[i] = hostTokens{th.token, replicas}
+		replicaRing = append(replicaRing, hostTokens{th.token, replicas})
 	}
 
-	if len(replicaRing) != len(tokens) {
+	if len(n.dcs) == len(dcRacks) && len(replicaRing) != len(tokens) {
 		panic(fmt.Sprintf("token map different size to token ring: got %d expected %d", len(replicaRing), len(tokens)))
 	}
 


Test patch:
diff --git a/topology_test.go b/topology_test.go
index cbc0f83..8c7acb7 100644
--- a/topology_test.go
+++ b/topology_test.go
@@ -90,76 +90,86 @@ func TestPlacementStrategy_NetworkStrategy(t *testing.T) {
 	}
 	sort.Sort(&tokenRing{tokens: tokens})
 
-	strat := &networkTopology{
-		dcs: map[string]int{
-			"dc1": 1,
-			"dc2": 2,
-			"dc3": 3,
+	strats := []*networkTopology{
+		&networkTopology{
+			dcs: map[string]int{
+				"dc1": 1,
+				"dc2": 2,
+				"dc3": 3,
+			},
+		},
+		&networkTopology{
+			dcs: map[string]int{
+				"dc2": 2,
+				"dc3": 3,
+			},
 		},
 	}
 
-	var expReplicas int
-	for _, rf := range strat.dcs {
-		expReplicas += rf
-	}
-
-	tokenReplicas := strat.replicaMap(&tokenRing{hosts: hosts, tokens: tokens})
-	if len(tokenReplicas) != len(tokens) {
-		t.Fatalf("expected replica map to have %d items but has %d", len(tokens), len(tokenReplicas))
-	}
-	if !sort.IsSorted(tokenReplicas) {
-		t.Fatal("replica map was not sorted by token")
-	}
+	for _, strat := range strats {
+		var expReplicas int
+		for _, rf := range strat.dcs {
+			expReplicas += rf
+		}
 
-	for token, replicas := range tokenReplicas {
-		if len(replicas.hosts) != expReplicas {
-			t.Fatalf("expected to have %d replicas got %d for token=%v", expReplicas, len(replicas.hosts), token)
+		tokenReplicas := strat.replicaMap(&tokenRing{hosts: hosts, tokens: tokens})
+		if needTokens := hostsPerDC * len(strat.dcs); len(tokenReplicas) != needTokens {
+			t.Fatalf("expected replica map to have %d items but has %d", needTokens, len(tokenReplicas))
+		}
+		if !sort.IsSorted(tokenReplicas) {
+			t.Fatal("replica map was not sorted by token")
 		}
-	}
 
-	for dc, rf := range strat.dcs {
-		dcTokens := dcRing[dc]
-		for i, th := range dcTokens {
-			token := th.token
-			allReplicas := tokenReplicas.replicasFor(token)
-			if allReplicas.token != token {
-				t.Fatalf("token %v not in replica map", token)
+		for token, replicas := range tokenReplicas {
+			if len(replicas.hosts) != expReplicas {
+				t.Fatalf("expected to have %d replicas got %d for token=%v", expReplicas, len(replicas.hosts), token)
 			}
+		}
 
-			var replicas []*HostInfo
-			for _, replica := range allReplicas.hosts {
-				if replica.dataCenter == dc {
-					replicas = append(replicas, replica)
+		for dc, rf := range strat.dcs {
+			dcTokens := dcRing[dc]
+			for i, th := range dcTokens {
+				token := th.token
+				allReplicas := tokenReplicas.replicasFor(token)
+				if allReplicas.token != token {
+					t.Fatalf("token %v not in replica map", token)
 				}
-			}
 
-			if len(replicas) != rf {
-				t.Fatalf("expected %d replicas in dc %q got %d", rf, dc, len(replicas))
-			}
+				var replicas []*HostInfo
+				for _, replica := range allReplicas.hosts {
+					if replica.dataCenter == dc {
+						replicas = append(replicas, replica)
+					}
+				}
+
+				if len(replicas) != rf {
+					t.Fatalf("expected %d replicas in dc %q got %d", rf, dc, len(replicas))
+				}
 
-			var lastRack string
-			for j, replica := range replicas {
-				// expected is in the next rack
-				var exp *HostInfo
-				if lastRack == "" {
-					// primary, first replica
-					exp = dcTokens[(i+j)%len(dcTokens)].host
-				} else {
-					for k := 0; k < len(dcTokens); k++ {
-						// walk around the ring from i + j to find the next host the
-						// next rack
-						p := (i + j + k) % len(dcTokens)
-						h := dcTokens[p].host
-						if h.rack != lastRack {
-							exp = h
-							break
+				var lastRack string
+				for j, replica := range replicas {
+					// expected is in the next rack
+					var exp *HostInfo
+					if lastRack == "" {
+						// primary, first replica
+						exp = dcTokens[(i+j)%len(dcTokens)].host
+					} else {
+						for k := 0; k < len(dcTokens); k++ {
+							// walk around the ring from i + j to find the next host the
+							// next rack
+							p := (i + j + k) % len(dcTokens)
+							h := dcTokens[p].host
+							if h.rack != lastRack {
+								exp = h
+								break
+							}
+						}
+						if exp.rack == lastRack {
+							panic("no more racks")
 						}
 					}
-					if exp.rack == lastRack {
-						panic("no more racks")
-					}
+					lastRack = replica.rack
 				}
-				lastRack = replica.rack
 			}
 		}
 	}