// checkBlpPositionList will ask the BinlogPlayerMap for its BlpPositionList,
// and check it contains one entry with the right data.
func checkBlpPositionList(t *testing.T, bpm *BinlogPlayerMap, vtClientSyncChannel chan *binlogplayer.VtClientMock) {
	// ask for BlpPositionList, make sure we got what we expect
	go func() {
		vtcm := binlogplayer.NewVtClientMock()
		vtcm.Result = &sqltypes.Result{
			Fields:       nil,
			RowsAffected: 1,
			InsertID:     0,
			Rows: [][]sqltypes.Value{
				{
					sqltypes.MakeString([]byte("MariaDB/0-1-1235")),
					sqltypes.MakeString([]byte("")),
				},
			},
		}
		vtClientSyncChannel <- vtcm
	}()
	bpl, err := bpm.BlpPositionList()
	if err != nil {
		t.Errorf("BlpPositionList failed: %v", err)
		return
	}
	if len(bpl) != 1 ||
		bpl[0].Uid != 1 ||
		bpl[0].Position != "MariaDB/0-1-1235" {
		t.Errorf("unexpected BlpPositionList: %v", bpl)
	}
}

// checkBlpPositionList will ask the BinlogPlayerMap for its BlpPositionList,
// and check it contains one entry with the right data.
func checkBlpPositionList(t *testing.T, bpm *BinlogPlayerMap, vtClientSyncChannel chan *binlogplayer.VtClientMock) {
	// ask for BlpPositionList, make sure we got what we expect
	go func() {
		vtcm := binlogplayer.NewVtClientMock()
		vtcm.Result = &mproto.QueryResult{
			Fields:       nil,
			RowsAffected: 1,
			InsertId:     0,
			Rows: [][]sqltypes.Value{
				[]sqltypes.Value{
					sqltypes.MakeString([]byte("MariaDB/0-1-1235")),
					sqltypes.MakeString([]byte("")),
				},
			},
		}
		vtClientSyncChannel <- vtcm
	}()
	bpl, err := bpm.BlpPositionList()
	if err != nil {
		t.Errorf("BlpPositionList failed: %v", err)
		return
	}
	if len(bpl.Entries) != 1 ||
		bpl.Entries[0].Uid != 1 ||
		bpl.Entries[0].Position.GTIDSet.(myproto.MariadbGTID).Domain != 0 ||
		bpl.Entries[0].Position.GTIDSet.(myproto.MariadbGTID).Server != 1 ||
		bpl.Entries[0].Position.GTIDSet.(myproto.MariadbGTID).Sequence != 1235 {
		t.Errorf("unexpected BlpPositionList: %v", bpl)
	}
}

//.........这里部分代码省略.........
	if ti.Type != topodatapb.TabletType_MASTER {
		t.Errorf("First health check failed to go to master: %v", ti.Type)
	}
	if !agent.QueryServiceControl.IsServing() {
		t.Errorf("Query service should be running")
	}
	if got := agent.QueryServiceControl.(*tabletservermock.Controller).CurrentTarget.TabletType; got != topodatapb.TabletType_MASTER {
		t.Errorf("invalid tabletserver target: got = %v, want = %v", got, topodatapb.TabletType_MASTER)
	}
	if _, err := expectBroadcastData(agent.QueryServiceControl, true, "", 20); err != nil {
		t.Fatal(err)
	}

	// Simulate a vertical split resharding where we set
	// SourceShards in the topo and enable filtered replication.
	_, err = agent.TopoServer.UpdateShardFields(ctx, "test_keyspace", "0", func(si *topo.ShardInfo) error {
		si.SourceShards = []*topodatapb.Shard_SourceShard{
			{
				Uid:      1,
				Keyspace: "source_keyspace",
				Shard:    "0",
				Tables: []string{
					"table1",
				},
			},
		}
		return nil
	})
	if err != nil {
		t.Fatalf("UpdateShardFields failed: %v", err)
	}

	// Mock out the BinlogPlayer client. Tell the BinlogPlayer not to start.
	vtClientMock := binlogplayer.NewVtClientMock()
	vtClientMock.AddResult(&sqltypes.Result{
		Fields:       nil,
		RowsAffected: 1,
		InsertID:     0,
		Rows: [][]sqltypes.Value{
			{
				sqltypes.MakeString([]byte("MariaDB/0-1-1234")),
				sqltypes.MakeString([]byte("DontStart")),
			},
		},
	})
	vtClientMocksChannel <- vtClientMock

	// Refresh the tablet state, as vtworker would do.
	// Since we change the QueryService state, we'll also trigger a health broadcast.
	agent.HealthReporter.(*fakeHealthCheck).reportReplicationDelay = 21 * time.Second
	agent.RefreshState(ctx)

	// (Destination) MASTER with enabled filtered replication mustn't serve anymore.
	if agent.QueryServiceControl.IsServing() {
		t.Errorf("Query service should not be running")
	}
	// Consume health broadcast sent out due to QueryService state change from
	// (MASTER, SERVING) to (MASTER, NOT_SERVING).
	// Since we didn't run healthcheck again yet, the broadcast data contains the
	// cached replication lag of 20 instead of 21.
	if bd, err := expectBroadcastData(agent.QueryServiceControl, false, "", 20); err == nil {
		if bd.RealtimeStats.BinlogPlayersCount != 1 {
			t.Fatalf("filtered replication must be enabled: %v", bd)
		}
	} else {
		t.Fatal(err)

// TestQueryServiceChangeImmediateHealthcheckResponse verifies that a change
// of the QueryService state or the tablet type will result into a broadcast
// of a StreamHealthResponse message.
func TestStateChangeImmediateHealthBroadcast(t *testing.T) {
	// BinlogPlayer will fail in the second retry because we don't fully mock
	// it. Retry faster to make it fail faster.
	flag.Set("binlog_player_retry_delay", "100ms")

	ctx := context.Background()
	agent, vtClientMocksChannel := createTestAgent(ctx, t)
	targetTabletType := topodatapb.TabletType_MASTER

	// Consume the first health broadcast triggered by ActionAgent.Start():
	//   (SPARE, SERVING) goes to (SPARE, NOT_SERVING).
	if _, err := expectBroadcastData(agent.QueryServiceControl, 0); err != nil {
		t.Fatal(err)
	}
	if err := expectStateChange(agent.QueryServiceControl, false, topodatapb.TabletType_SPARE); err != nil {
		t.Fatal(err)
	}

	// Run health check to get changed from SPARE to MASTER.
	agent.HealthReporter.(*fakeHealthCheck).reportReplicationDelay = 20 * time.Second
	agent.runHealthCheck(targetTabletType)
	ti, err := agent.TopoServer.GetTablet(ctx, tabletAlias)
	if err != nil {
		t.Fatalf("GetTablet failed: %v", err)
	}
	if ti.Type != targetTabletType {
		t.Errorf("First health check failed to go to replica: %v", ti.Type)
	}
	if !agent.QueryServiceControl.IsServing() {
		t.Errorf("Query service should be running")
	}
	if got := agent.QueryServiceControl.(*tabletservermock.Controller).CurrentTarget.TabletType; got != targetTabletType {
		t.Errorf("invalid tabletserver target: got = %v, want = %v", got, targetTabletType)
	}
	if _, err := expectBroadcastData(agent.QueryServiceControl, 20); err != nil {
		t.Fatal(err)
	}
	if err := expectStateChange(agent.QueryServiceControl, true, targetTabletType); err != nil {
		t.Fatal(err)
	}

	// Simulate a vertical split resharding where we set SourceShards in the topo
	// and enable filtered replication.
	si, err := agent.TopoServer.GetShard(ctx, "test_keyspace", "0")
	if err != nil {
		t.Fatalf("GetShard failed: %v", err)
	}
	si.SourceShards = []*topodatapb.Shard_SourceShard{
		{
			Uid:      1,
			Keyspace: "source_keyspace",
			Shard:    "0",
			Tables: []string{
				"table1",
			},
		},
	}
	if err := agent.TopoServer.UpdateShard(ctx, si); err != nil {
		t.Fatalf("UpdateShard failed: %v", err)
	}
	// Mock out the BinlogPlayer client. Tell the BinlogPlayer not to start.
	vtClientMock := binlogplayer.NewVtClientMock()
	vtClientMock.Result = &sqltypes.Result{
		Fields:       nil,
		RowsAffected: 1,
		InsertID:     0,
		Rows: [][]sqltypes.Value{
			{
				sqltypes.MakeString([]byte("MariaDB/0-1-1234")),
				sqltypes.MakeString([]byte("DontStart")),
			},
		},
	}
	vtClientMocksChannel <- vtClientMock

	// Refresh the tablet state, as vtworker would do.
	// Since we change the QueryService state, we'll also trigger a health broadcast.
	agent.HealthReporter.(*fakeHealthCheck).reportReplicationDelay = 21 * time.Second
	agent.RPCWrapLockAction(ctx, actionnode.TabletActionRefreshState, "", "", true, func() error {
		agent.RefreshState(ctx)
		return nil
	})
	// (Destination) MASTER with enabled filtered replication mustn't serve anymore.
	if agent.QueryServiceControl.IsServing() {
		t.Errorf("Query service should not be running")
	}
	// Consume health broadcast sent out due to QueryService state change from
	// (MASTER, SERVING) to (MASTER, NOT_SERVING).
	// Since we didn't run healthcheck again yet, the broadcast data contains the
	// cached replication lag of 20 instead of 21.
	if bd, err := expectBroadcastData(agent.QueryServiceControl, 20); err == nil {
		if bd.RealtimeStats.BinlogPlayersCount != 1 {
			t.Fatalf("filtered replication must be enabled: %v", bd)
		}
	} else {
		t.Fatal(err)
	}
//.........这里部分代码省略.........

//.........这里部分代码省略.........
		},
		Keyspace: "destination",
		Shard:    "0",
	}

	si, err := ts.GetShard(ctx, "destination", "0")
	if err != nil {
		t.Fatalf("GetShard failed: %v", err)
	}
	si.SourceShards = []*topodatapb.Shard_SourceShard{
		{
			Uid:      1,
			Keyspace: "source",
			Shard:    "0",
			Tables: []string{
				"table1",
				"funtables_*",
			},
		},
	}
	if err := ts.UpdateShard(ctx, si); err != nil {
		t.Fatalf("UpdateShard failed: %v", err)
	}

	// now we have a source, adding players
	bpm.RefreshMap(ctx, tablet, si)
	if !bpm.isRunningFilteredReplication() {
		t.Errorf("isRunningFilteredReplication should be true")
	}

	// write a mocked vtClientMock that will be used to read the
	// start position at first. Note this also synchronizes the player,
	// so we can then check mysqlDaemon.BinlogPlayerEnabled.
	vtClientMock := binlogplayer.NewVtClientMock()
	vtClientMock.Result = &sqltypes.Result{
		Fields:       nil,
		RowsAffected: 1,
		InsertID:     0,
		Rows: [][]sqltypes.Value{
			{
				sqltypes.MakeString([]byte("MariaDB/0-1-1234")),
				sqltypes.MakeString([]byte("")),
			},
		},
	}
	vtClientSyncChannel <- vtClientMock
	if !mysqlDaemon.BinlogPlayerEnabled {
		t.Errorf("mysqlDaemon.BinlogPlayerEnabled should be true")
	}

	// the client will then try to connect to the remote tablet.
	// give it what it needs.
	fbc := newFakeBinlogClient(t, 100)
	fbc.expectedTables = "table1,funtables_one"
	clientSyncChannel <- fbc

	// now we can feed an event through the fake connection
	vtClientMock.CommitChannel = make(chan []string)
	fbc.tablesChannel <- &binlogdatapb.BinlogTransaction{
		Statements: []*binlogdatapb.BinlogTransaction_Statement{
			{
				Category: binlogdatapb.BinlogTransaction_Statement_BL_DML,
				Sql:      "INSERT INTO tablet VALUES(1)",
			},
		},
		Timestamp:     72,

func TestBinlogPlayerMapHorizontalSplitStopStartUntil(t *testing.T) {
	ts := zktestserver.New(t, []string{"cell1"})
	ctx := context.Background()

	// create the keyspace, a full set of covering shards,
	// and a new split destination shard.
	if err := ts.CreateKeyspace(ctx, "ks", &topodatapb.Keyspace{
		ShardingColumnType: topodatapb.KeyspaceIdType_UINT64,
		ShardingColumnName: "sharding_key",
	}); err != nil {
		t.Fatalf("CreateKeyspace failed: %v", err)
	}
	for _, shard := range []string{"-80", "80-", "40-60"} {
		if err := ts.CreateShard(ctx, "ks", shard); err != nil {
			t.Fatalf("CreateShard failed: %v", err)
		}
	}

	// create one replica remote tablet in source shard, we will
	// use it as a source for filtered replication.
	createSourceTablet(t, "test_horizontal_until", ts, "ks", "-80")

	// register a binlog player factory that will return the instances
	// we want
	clientSyncChannel := make(chan *fakeBinlogClient)
	binlogplayer.RegisterClientFactory("test_horizontal_until", func() binlogplayer.Client {
		return <-clientSyncChannel
	})
	flag.Lookup("binlog_player_protocol").Value.Set("test_horizontal_until")

	// create the BinlogPlayerMap on the local tablet
	// (note that local tablet is never in the topology, we don't
	// need it there at all)
	mysqlDaemon := &mysqlctl.FakeMysqlDaemon{MysqlPort: 3306}
	vtClientSyncChannel := make(chan *binlogplayer.VtClientMock)
	bpm := NewBinlogPlayerMap(ts, mysqlDaemon, func() binlogplayer.VtClient {
		return <-vtClientSyncChannel
	})

	tablet := &topodatapb.Tablet{
		Alias: &topodatapb.TabletAlias{
			Cell: "cell1",
			Uid:  1,
		},
		KeyRange: &topodatapb.KeyRange{
			Start: []byte{0x40},
			End:   []byte{0x60},
		},
		Keyspace: "ks",
		Shard:    "40-60",
	}

	si, err := ts.GetShard(ctx, "ks", "40-60")
	if err != nil {
		t.Fatalf("GetShard failed: %v", err)
	}
	si.SourceShards = []*topodatapb.Shard_SourceShard{
		{
			Uid:      1,
			Keyspace: "ks",
			Shard:    "-80",
			KeyRange: &topodatapb.KeyRange{
				End: []byte{0x80},
			},
		},
	}
	if err := ts.UpdateShard(ctx, si); err != nil {
		t.Fatalf("UpdateShard failed: %v", err)
	}

	// now we have a source, adding players
	bpm.RefreshMap(ctx, tablet, si)
	if !bpm.isRunningFilteredReplication() {
		t.Errorf("isRunningFilteredReplication should be true")
	}

	// write a mocked vtClientMock that will be used to read the
	// start position at first. Note this also synchronizes the player,
	// so we can then check mysqlDaemon.BinlogPlayerEnabled.
	vtClientMock := binlogplayer.NewVtClientMock()
	vtClientMock.Result = &sqltypes.Result{
		Fields:       nil,
		RowsAffected: 1,
		InsertID:     0,
		Rows: [][]sqltypes.Value{
			{
				sqltypes.MakeString([]byte("MariaDB/0-1-1234")),
				sqltypes.MakeString([]byte("")),
			},
		},
	}
	vtClientSyncChannel <- vtClientMock
	if !mysqlDaemon.BinlogPlayerEnabled {
		t.Errorf("mysqlDaemon.BinlogPlayerEnabled should be true")
	}

	// the client will then try to connect to the remote tablet.
	// give it what it needs.
	fbc := newFakeBinlogClient(t, 100)
	fbc.expectedKeyRange = "40-60"
//.........这里部分代码省略.........