Changes after self review

Signed-off-by: Matt Lord <mattalord@gmail.com>

go/test/endtoend/vreplication/vstream_test.go

@@ -538,7 +538,7 @@ func TestMultiVStreamsKeyspaceReshard(t *testing.T) {
 	_, err := vc.AddKeyspace(t, []*Cell{defaultCell}, "global", "0", vschemaUnsharded, schemaUnsharded, defaultReplicas, defaultRdonly, baseTabletID, nil)
 	require.NoError(t, err)
 
-	// Setup the keyspace with our old shards.
+	// Setup the keyspace with our old/original shards.
 	keyspace, err := vc.AddKeyspace(t, []*Cell{defaultCell}, ks, oldShards, vschemaSharded, schemaSharded, defaultReplicas, defaultRdonly, baseTabletID+1000, nil)
 	require.NoError(t, err)
 
@@ -553,23 +553,12 @@ func TestMultiVStreamsKeyspaceReshard(t *testing.T) {
 	vstreamConn, err := vtgateconn.Dial(ctx, fmt.Sprintf("%s:%d", vc.ClusterConfig.hostname, vc.ClusterConfig.vtgateGrpcPort))
 	require.NoError(t, err)
 	defer vstreamConn.Close()
-	vgtid := &binlogdatapb.VGtid{
-		ShardGtids: []*binlogdatapb.ShardGtid{{
-			Keyspace: "/.*", // Match all keyspaces just to be more realistic.
-		}}}
-
-	filter := &binlogdatapb.Filter{
-		Rules: []*binlogdatapb.Rule{{
-			// Only stream the customer table.
-			Match: "customer",
-		}},
-	}
-	flags := &vtgatepb.VStreamFlags{}
 
 	// Ensure that we're starting with a clean slate.
 	_, err = vtgateConn.ExecuteFetch(fmt.Sprintf("delete from %s.customer", ks), 1000, false)
 	require.NoError(t, err)
 
+	// Coordinate go-routines.
 	streamCtx, streamCancel := context.WithTimeout(ctx, 1*time.Minute)
 	defer streamCancel()
 	done := make(chan struct{})
@@ -598,6 +587,19 @@ func TestMultiVStreamsKeyspaceReshard(t *testing.T) {
 	reshardAction(t, "Create", wf, ks, oldShards, newShards, defaultCellName, tabletType)
 	waitForWorkflowState(t, vc, fmt.Sprintf("%s.%s", ks, wf), binlogdatapb.VReplicationWorkflowState_Running.String())
 
+	vgtid := &binlogdatapb.VGtid{
+		ShardGtids: []*binlogdatapb.ShardGtid{{
+			Keyspace: "/.*", // Match all keyspaces just to be more realistic.
+		}}}
+
+	filter := &binlogdatapb.Filter{
+		Rules: []*binlogdatapb.Rule{{
+			// Only stream the customer table and its sequence backing table.
+			Match: "/customer.*",
+		}},
+	}
+	flags := &vtgatepb.VStreamFlags{}
+
 	// Stream events but stop once we have a VGTID with positions for the old/original shards.
 	var newVGTID *binlogdatapb.VGtid
 	func() {
@@ -639,7 +641,7 @@ func TestMultiVStreamsKeyspaceReshard(t *testing.T) {
 					}
 				}
 			default:
-				require.FailNow(t, "VStream returned unexpected error: %v", err)
+				require.FailNow(t, fmt.Sprintf("VStream returned unexpected error: %v", err))
 				return
 			}
 			select {
@@ -650,6 +652,7 @@ func TestMultiVStreamsKeyspaceReshard(t *testing.T) {
 		}
 	}()
 
+	// Confirm that we have shard GTIDs for the global shard and the old/original shards.
 	require.Len(t, newVGTID.GetShardGtids(), 3)
 
 	// Switch the traffic to the new shards.
@@ -695,14 +698,15 @@ func TestMultiVStreamsKeyspaceReshard(t *testing.T) {
 		}
 	}()
 
-	require.GreaterOrEqual(t, oldShardRowEvents, 1)
-	require.GreaterOrEqual(t, newShardRowEvents, 1)
+	// We should have a mix of events across the old and new shards.
+	require.NotZero(t, oldShardRowEvents)
+	require.NotZero(t, newShardRowEvents)
 
 	// The number of row events streamed by the VStream API should match the number of rows inserted.
 	customerResult := execVtgateQuery(t, vtgateConn, ks, "select count(*) from customer")
-	insertedCustomerRows, err := customerResult.Rows[0][0].ToCastInt64()
+	customerCount, err := customerResult.Rows[0][0].ToInt64()
 	require.NoError(t, err)
-	require.Equal(t, insertedCustomerRows, int64(oldShardRowEvents+newShardRowEvents))
+	require.Equal(t, customerCount, int64(oldShardRowEvents+newShardRowEvents))
 }
 
 func TestVStreamFailover(t *testing.T) {

go/vt/vtgate/vstream_manager.go

@@ -940,30 +940,33 @@ func (vs *vstream) getJournalEvent(ctx context.Context, sgtid *binlogdatapb.Shar
 }
 
 // keyspaceHasBeenResharded returns true if the keyspace's serving shard set has changed
-// since the last VStream as indicated by the shard definitions provided in the vgtid.
+// since the last VStream as indicated by the shard definitions provided in the VGTID.
 func (vs *vstream) keyspaceHasBeenResharded(ctx context.Context, keyspace string) (bool, error) {
 	shards, err := vs.ts.FindAllShardsInKeyspace(ctx, keyspace, nil)
 	if err != nil || len(shards) == 0 {
 		return false, err
 	}
 
-	// First check the typical case, where the vgtid shards match the serving shards.
+	// First check the typical case, where the VGTID shards match the serving shards.
 	// In that case it's NOT possible that an applicable reshard has happened because
-	// the vgtid contains shards that are all serving.
+	// the VGTID contains shards that are all serving.
 	reshardPossible := false
 	ksShardGTIDs := make([]*binlogdatapb.ShardGtid, 0, len(vs.vgtid.ShardGtids))
-	for _, g := range vs.vgtid.ShardGtids {
-		if g.GetKeyspace() == keyspace {
-			ksShardGTIDs = append(ksShardGTIDs, g)
+	for _, s := range vs.vgtid.ShardGtids {
+		if s.GetKeyspace() == keyspace {
+			ksShardGTIDs = append(ksShardGTIDs, s)
 		}
 	}
 	for _, s := range ksShardGTIDs {
-		if !shards[s.GetShard()].GetIsPrimaryServing() {
+		shard := shards[s.GetShard()]
+		if shard == nil {
+			return false, vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "shard provided in VGTID, %s, not found in keyspace %s", s.GetShard(), keyspace)
+		}
+		if !shard.GetIsPrimaryServing() {
 			reshardPossible = true
 			break
 		}
 	}
-	log.Errorf("DEBUG: reshard possible: %v", reshardPossible)
 	if !reshardPossible {
 		return false, nil
 	}