func TestResolverDmlOnMultipleKeyspaceIds(t *testing.T) {
	kid10, err := key.HexKeyspaceId("10").Unhex()
	if err != nil {
		t.Errorf("Error encoding keyspace id")
	}
	kid25, err := key.HexKeyspaceId("25").Unhex()
	if err != nil {
		t.Errorf("Error encoding keyspace id")
	}
	res := NewResolver(new(sandboxTopo), "", "aa", retryDelay, 0, connTimeoutTotal, connTimeoutPerConn, connLife)

	s := createSandbox("TestResolverDmlOnMultipleKeyspaceIds")
	sbc0 := &sandboxConn{}
	s.MapTestConn("-20", sbc0)
	sbc1 := &sandboxConn{}
	s.MapTestConn("20-40", sbc1)

	errStr := "DML should not span multiple keyspace_ids"
	_, err = res.ExecuteKeyspaceIds(context.Background(),
		"update table set a = b",
		nil,
		"TestResolverExecuteKeyspaceIds",
		[]key.KeyspaceId{kid10, kid25},
		pb.TabletType_MASTER,
		nil,
		false)
	if err == nil {
		t.Errorf("want %v, got nil", errStr)
	}
}

// ValidateShardName takes a shard name and sanitizes it, and also returns
// the KeyRange.
func ValidateShardName(shard string) (string, key.KeyRange, error) {
	if !strings.Contains(shard, "-") {
		return shard, key.KeyRange{}, nil
	}

	parts := strings.Split(shard, "-")
	if len(parts) != 2 {
		return "", key.KeyRange{}, fmt.Errorf("Invalid shardId, can only contain one '-': %v", shard)
	}

	start, err := key.HexKeyspaceId(parts[0]).Unhex()
	if err != nil {
		return "", key.KeyRange{}, err
	}

	end, err := key.HexKeyspaceId(parts[1]).Unhex()
	if err != nil {
		return "", key.KeyRange{}, err
	}

	if end != key.MaxKey && start >= end {
		return "", key.KeyRange{}, fmt.Errorf("Out of order keys: %v is not strictly smaller than %v", start, end)
	}

	return strings.ToUpper(shard), key.KeyRange{Start: start, End: end}, nil
}

func TestResolverExecuteBatchKeyspaceIds(t *testing.T) {
	testResolverGeneric(t, "TestResolverExecuteBatchKeyspaceIds", func() (*mproto.QueryResult, error) {
		kid10, err := key.HexKeyspaceId("10").Unhex()
		if err != nil {
			return nil, err
		}
		kid25, err := key.HexKeyspaceId("25").Unhex()
		if err != nil {
			return nil, err
		}
		query := &proto.KeyspaceIdBatchQuery{
			Queries: []proto.BoundKeyspaceIdQuery{{
				Sql:           "query",
				BindVariables: nil,
				Keyspace:      "TestResolverExecuteBatchKeyspaceIds",
				KeyspaceIds:   []key.KeyspaceId{kid10, kid25},
			}},
			TabletType:    topo.TYPE_MASTER,
			AsTransaction: false,
		}
		res := NewResolver(new(sandboxTopo), "", "aa", 1*time.Millisecond, 0, 2*time.Millisecond, 1*time.Millisecond, 24*time.Hour)
		qrs, err := res.ExecuteBatchKeyspaceIds(context.Background(), query)
		if err != nil {
			return nil, err
		}
		return &qrs.List[0], err
	})
}

func TestResolverExecuteEntityIds(t *testing.T) {
	testResolverGeneric(t, "TestResolverExecuteEntityIds", func() (*mproto.QueryResult, error) {
		kid10, err := key.HexKeyspaceId("10").Unhex()
		if err != nil {
			return nil, err
		}
		kid25, err := key.HexKeyspaceId("25").Unhex()
		if err != nil {
			return nil, err
		}
		res := NewResolver(new(sandboxTopo), "", "aa", 1*time.Millisecond, 0, 2*time.Millisecond, 1*time.Millisecond, 24*time.Hour)
		return res.ExecuteEntityIds(context.Background(),
			"query",
			nil,
			"TestResolverExecuteEntityIds",
			"col",
			[]proto.EntityId{
				proto.EntityId{
					ExternalID: 0,
					KeyspaceID: kid10,
				},
				proto.EntityId{
					ExternalID: "1",
					KeyspaceID: kid25,
				},
			},
			pb.TabletType_MASTER,
			nil,
			false)
	})
}

func TestBoundKeyspaceIdQueriesToBoundShardQueries(t *testing.T) {
	ts := new(sandboxTopo)
	kid10, err := key.HexKeyspaceId("10").Unhex()
	if err != nil {
		t.Error(err)
	}
	kid25, err := key.HexKeyspaceId("25").Unhex()
	if err != nil {
		t.Error(err)
	}
	var testCases = []struct {
		idQueries    []proto.BoundKeyspaceIdQuery
		shardQueries []proto.BoundShardQuery
	}{
		{
			idQueries: []proto.BoundKeyspaceIdQuery{
				{
					Sql:           "q1",
					BindVariables: map[string]interface{}{"q1var": 1},
					Keyspace:      KsTestSharded,
					KeyspaceIds:   []key.KeyspaceId{kid10, kid25},
				}, {
					Sql:           "q2",
					BindVariables: map[string]interface{}{"q2var": 2},
					Keyspace:      KsTestSharded,
					KeyspaceIds:   []key.KeyspaceId{kid25, kid25},
				},
			},
			shardQueries: []proto.BoundShardQuery{
				{
					Sql:           "q1",
					BindVariables: map[string]interface{}{"q1var": 1},
					Keyspace:      KsTestSharded,
					Shards:        []string{"-20", "20-40"},
				}, {
					Sql:           "q2",
					BindVariables: map[string]interface{}{"q2var": 2},
					Keyspace:      KsTestSharded,
					Shards:        []string{"20-40"},
				},
			},
		},
	}

	for _, testCase := range testCases {
		shardQueries, err := boundKeyspaceIDQueriesToBoundShardQueries(context.Background(), ts, "", pb.TabletType_MASTER, testCase.idQueries)
		if err != nil {
			t.Error(err)
		}
		// Sort shards, because they're random otherwise.
		for _, shardQuery := range shardQueries {
			sort.Strings(shardQuery.Shards)
		}
		if !reflect.DeepEqual(testCase.shardQueries, shardQueries) {
			got, _ := json.Marshal(shardQueries)
			want, _ := json.Marshal(testCase.shardQueries)
			t.Errorf("idQueries: %#v\nResponse:   %s\nExepecting: %s", testCase.idQueries, got, want)
		}
	}
}

func TestResolverDmlOnMultipleKeyspaceIds(t *testing.T) {
	kid10, err := key.HexKeyspaceId("10").Unhex()
	if err != nil {
		t.Errorf("Error encoding keyspace id")
	}
	kid25, err := key.HexKeyspaceId("25").Unhex()
	if err != nil {
		t.Errorf("Error encoding keyspace id")
	}
	query := &proto.KeyspaceIdQuery{
		Sql:         "update table set a = b",
		Keyspace:    "TestResolverExecuteKeyspaceIds",
		KeyspaceIds: []key.KeyspaceId{kid10, kid25},
		TabletType:  topo.TYPE_MASTER,
	}
	res := NewResolver(new(sandboxTopo), "", "aa", 1*time.Millisecond, 0, 2*time.Millisecond, 1*time.Millisecond, 24*time.Hour)

	s := createSandbox("TestResolverDmlOnMultipleKeyspaceIds")
	sbc0 := &sandboxConn{}
	s.MapTestConn("-20", sbc0)
	sbc1 := &sandboxConn{}
	s.MapTestConn("20-40", sbc1)

	errStr := "DML should not span multiple keyspace_ids"
	_, err = res.ExecuteKeyspaceIds(context.Background(), query)
	if err == nil {
		t.Errorf("want %v, got nil", errStr)
	}
}

func TestResolverExecuteEntityIds(t *testing.T) {
	testResolverGeneric(t, "TestResolverExecuteEntityIds", func() (*mproto.QueryResult, error) {
		kid10, err := key.HexKeyspaceId("10").Unhex()
		if err != nil {
			return nil, err
		}
		kid25, err := key.HexKeyspaceId("25").Unhex()
		if err != nil {
			return nil, err
		}
		query := &proto.EntityIdsQuery{
			Sql:              "query",
			Keyspace:         "TestResolverExecuteEntityIds",
			EntityColumnName: "col",
			EntityKeyspaceIDs: []proto.EntityId{
				proto.EntityId{
					ExternalID: 0,
					KeyspaceID: kid10,
				},
				proto.EntityId{
					ExternalID: "1",
					KeyspaceID: kid25,
				},
			},
			TabletType: topo.TYPE_MASTER,
		}
		res := NewResolver(new(sandboxTopo), "", "aa", 1*time.Millisecond, 0, 2*time.Millisecond, 1*time.Millisecond, 24*time.Hour)
		return res.ExecuteEntityIds(context.Background(), query)
	})
}

func TestResolverExecuteBatchKeyspaceIds(t *testing.T) {
	testResolverGeneric(t, "TestResolverExecuteBatchKeyspaceIds", func() (*mproto.QueryResult, error) {
		kid10, err := key.HexKeyspaceId("10").Unhex()
		if err != nil {
			return nil, err
		}
		kid25, err := key.HexKeyspaceId("25").Unhex()
		if err != nil {
			return nil, err
		}
		res := NewResolver(nil, topo.Server{}, new(sandboxTopo), "", "aa", retryDelay, 0, connTimeoutTotal, connTimeoutPerConn, connLife, "")
		qrs, err := res.ExecuteBatchKeyspaceIds(context.Background(),
			[]proto.BoundKeyspaceIdQuery{{
				Sql:           "query",
				BindVariables: nil,
				Keyspace:      "TestResolverExecuteBatchKeyspaceIds",
				KeyspaceIds:   []key.KeyspaceId{kid10, kid25},
			}},
			pb.TabletType_MASTER,
			false,
			nil)
		if err != nil {
			return nil, err
		}
		return &qrs.List[0], err
	})
}

func NewBinlogPlayer(dbClient VtClient, startPosition *binlogRecoveryState, tables []string, txnBatch int, maxTxnInterval time.Duration, execDdl bool) (*BinlogPlayer, error) {
	if !startPositionValid(startPosition) {
		log.Fatalf("Invalid Start Position")
	}

	blp := new(BinlogPlayer)
	blp.recoveryState = *startPosition

	// convert start and end keyrange
	var err error
	blp.keyrange.Start, err = key.HexKeyspaceId(startPosition.KeyrangeStart).Unhex()
	if err != nil {
		return nil, fmt.Errorf("Error in Unhex for %v, '%v'", startPosition.KeyrangeStart, err)
	}
	blp.keyrange.End, err = key.HexKeyspaceId(startPosition.KeyrangeEnd).Unhex()
	if err != nil {
		return nil, fmt.Errorf("Error in Unhex for %v, '%v'", startPosition.KeyrangeEnd, err)
	}
	blp.txnIndex = 0
	blp.inTxn = false
	blp.txnBuffer = make([]*cproto.BinlogResponse, 0, MAX_TXN_BATCH)
	blp.dbClient = dbClient
	blp.tables = tables
	blp.blplStats = NewBlplStats()
	blp.batchStart = time.Now()
	blp.txnBatch = txnBatch
	blp.maxTxnInterval = maxTxnInterval
	blp.execDdl = execDdl
	return blp, nil
}

func (blServer *BinlogServer) ServeBinlog(req *mysqlctl.BinlogServerRequest, sendReply mysqlctl.SendUpdateStreamResponse) error {
	defer func() {
		if x := recover(); x != nil {
			//Send the error to the client.
			_, ok := x.(*BinlogParseError)
			if !ok {
				relog.Error("Uncaught panic at top-most level: '%v'", x)
				//panic(x)
			}
			sendError(sendReply, req.StartPosition.String(), x.(error), nil)
		}
	}()

	relog.Info("received req: %v kr start %v end %v", req.StartPosition.String(), req.KeyspaceStart, req.KeyspaceEnd)
	if !isRequestValid(req) {
		panic(NewBinlogParseError("Invalid request, cannot serve the stream"))
	}

	usingRelayLogs := false
	var binlogPrefix, logsDir string
	if req.StartPosition.RelayFilename != "" {
		usingRelayLogs = true
		binlogPrefix = blServer.mycnf.RelayLogPath
		logsDir = path.Dir(binlogPrefix)
		if !mysqlctl.IsRelayPositionValid(&req.StartPosition, logsDir) {
			panic(NewBinlogParseError(fmt.Sprintf("Invalid start position %v, cannot serve the stream, cannot locate start position", req.StartPosition)))
		}
	} else {
		binlogPrefix = blServer.mycnf.BinLogPath
		logsDir = path.Dir(binlogPrefix)
		if !mysqlctl.IsMasterPositionValid(&req.StartPosition) {
			panic(NewBinlogParseError(fmt.Sprintf("Invalid start position %v, cannot serve the stream, cannot locate start position", req.StartPosition)))
		}
	}

	startKey, err := key.HexKeyspaceId(req.KeyspaceStart).Unhex()
	if err != nil {
		panic(NewBinlogParseError(fmt.Sprintf("Unhex on key '%v' failed", req.KeyspaceStart)))
	}
	endKey, err := key.HexKeyspaceId(req.KeyspaceEnd).Unhex()
	if err != nil {
		panic(NewBinlogParseError(fmt.Sprintf("Unhex on key '%v' failed", req.KeyspaceEnd)))
	}
	keyRange := &key.KeyRange{Start: startKey, End: endKey}

	blp := NewBlp(&req.StartPosition, blServer, keyRange)
	blp.usingRelayLogs = usingRelayLogs
	blp.binlogPrefix = binlogPrefix
	blp.logMetadata = mysqlctl.NewSlaveMetadata(logsDir, blServer.mycnf.RelayLogInfoPath)

	relog.Info("usingRelayLogs %v blp.binlogPrefix %v logsDir %v", blp.usingRelayLogs, blp.binlogPrefix, logsDir)
	blp.streamBinlog(sendReply)
	return nil
}

func TestVTGateExecuteBatchKeyspaceIds(t *testing.T) {
	//TODO(sougou): Fix test
	t.Skip()
	s := createSandbox("TestVTGateExecuteBatchKeyspaceIds")
	s.MapTestConn("-20", &sandboxConn{})
	s.MapTestConn("20-40", &sandboxConn{})
	kid10, err := key.HexKeyspaceId("10").Unhex()
	if err != nil {
		t.Errorf("want nil, got %v", err)
	}
	kid30, err := key.HexKeyspaceId("30").Unhex()
	if err != nil {
		t.Errorf("want nil, got %v", err)
	}
	q := proto.KeyspaceIdBatchQuery{
		Queries: []proto.BoundKeyspaceIdQuery{{
			Sql:           "query",
			BindVariables: nil,
			Keyspace:      "TestVTGateExecuteBatchKeyspaceIds",
			KeyspaceIds:   []key.KeyspaceId{kid10, kid30},
		}, {
			Sql:           "query",
			BindVariables: nil,
			Keyspace:      "TestVTGateExecuteBatchKeyspaceIds",
			KeyspaceIds:   []key.KeyspaceId{kid10, kid30},
		}},
		TabletType: topo.TYPE_MASTER,
	}
	qrl := new(proto.QueryResultList)
	err = rpcVTGate.ExecuteBatchKeyspaceIds(context.Background(), &q, qrl)
	if err != nil {
		t.Errorf("want nil, got %v", err)
	}
	if len(qrl.List) != 2 {
		t.Errorf("want 2, got %v", len(qrl.List))
	}
	if qrl.List[0].RowsAffected != 2 {
		t.Errorf("want 2, got %v", qrl.List[0].RowsAffected)
	}
	if qrl.Session != nil {
		t.Errorf("want nil, got %+v\n", qrl.Session)
	}

	q.Session = new(proto.Session)
	rpcVTGate.Begin(context.Background(), q.Session)
	rpcVTGate.ExecuteBatchKeyspaceIds(context.Background(), &q, qrl)
	if len(q.Session.ShardSessions) != 2 {
		t.Errorf("want 2, got %d", len(q.Session.ShardSessions))
	}
}

func hki(hexValue string) key.KeyspaceId {
	k, err := key.HexKeyspaceId(hexValue).Unhex()
	if err != nil {
		panic(err)
	}
	return k
}

func hexOrDie(t *testing.T, hex string) key.KeyspaceId {
	kid, err := key.HexKeyspaceId(hex).Unhex()
	if err != nil {
		t.Fatalf("Unhex failed: %v", err)
	}
	return kid
}

func TestResolverStreamExecuteKeyRanges(t *testing.T) {
	kid10, err := key.HexKeyspaceId("10").Unhex()
	if err != nil {
		t.Errorf(err.Error())
	}
	kid15, err := key.HexKeyspaceId("15").Unhex()
	if err != nil {
		t.Errorf(err.Error())
	}
	kid25, err := key.HexKeyspaceId("25").Unhex()
	if err != nil {
		t.Errorf(err.Error())
	}
	createSandbox("TestResolverStreamExecuteKeyRanges")
	// streaming a single shard
	testResolverStreamGeneric(t, "TestResolverStreamExecuteKeyRanges", func() (*mproto.QueryResult, error) {
		res := NewResolver(new(sandboxTopo), "", "aa", 1*time.Millisecond, 0, 2*time.Millisecond, 1*time.Millisecond, 24*time.Hour)
		qr := new(mproto.QueryResult)
		err = res.StreamExecuteKeyRanges(context.Background(),
			"query",
			nil,
			"TestResolverStreamExecuteKeyRanges",
			[]key.KeyRange{key.KeyRange{Start: kid10, End: kid15}},
			pb.TabletType_MASTER,
			func(r *mproto.QueryResult) error {
				appendResult(qr, r)
				return nil
			})
		return qr, err
	})
	// streaming multiple shards
	testResolverStreamGeneric(t, "TestResolverStreamExecuteKeyRanges", func() (*mproto.QueryResult, error) {
		res := NewResolver(new(sandboxTopo), "", "aa", 1*time.Millisecond, 0, 2*time.Millisecond, 1*time.Millisecond, 24*time.Hour)
		qr := new(mproto.QueryResult)
		err = res.StreamExecuteKeyRanges(context.Background(),
			"query",
			nil,
			"TestResolverStreamExecuteKeyRanges",
			[]key.KeyRange{key.KeyRange{Start: kid10, End: kid25}},
			pb.TabletType_MASTER,
			func(r *mproto.QueryResult) error {
				appendResult(qr, r)
				return nil
			})
		return qr, err
	})
}

func multiRestoreCmd(mysqld *mysqlctl.Mysqld, subFlags *flag.FlagSet, args []string) {
	start := subFlags.String("start", "", "start of the key range")
	end := subFlags.String("end", "", "end of the key range")
	fetchRetryCount := subFlags.Int("fetch-retry-count", 3, "how many times to retry a failed transfer")
	concurrency := subFlags.Int("concurrency", 8, "how many concurrent db inserts to run simultaneously")
	fetchConcurrency := subFlags.Int("fetch-concurrency", 4, "how many files to fetch simultaneously")
	insertTableConcurrency := subFlags.Int("insert-table-concurrency", 4, "how many myisam tables to load into a single destination table simultaneously")
	strategy := subFlags.String("strategy", "", "which strategy to use for restore, can contain:\n"+
		"    skipAutoIncrement(TTT): we won't add the AUTO_INCREMENT back to that table\n"+
		"    delayPrimaryKey: we won't add the primary key until after the table is populated\n"+
		"    delaySecondaryIndexes: we won't add the secondary indexes until after the table is populated\n"+
		"    useMyIsam: create the table as MyISAM, then convert it to InnoDB after population\n"+
		"    writeBinLogs: write all operations to the binlogs")

	subFlags.Parse(args)

	s, err := key.HexKeyspaceId(*start).Unhex()
	if err != nil {
		relog.Fatal("Invalid start key %v: %v", *start, err)
	}
	e, err := key.HexKeyspaceId(*end).Unhex()
	if err != nil {
		relog.Fatal("Invalid end key %v: %v", *end, err)
	}
	keyRange := key.KeyRange{Start: s, End: e}

	if subFlags.NArg() < 2 {
		relog.Fatal("multirestore requires <destination_dbname> <source_host>[/<source_dbname>]... %v", args)
	}
	dbName, dbis := subFlags.Arg(0), subFlags.Args()[1:]
	sources := make([]*url.URL, len(dbis))
	uids := make([]uint32, len(dbis))
	for i, dbi := range dbis {
		if !strings.HasPrefix(dbi, "vttp://") && !strings.HasPrefix(dbi, "http://") {
			dbi = "vttp://" + dbi
		}
		dbUrl, err := url.Parse(dbi)
		if err != nil {
			relog.Fatal("incorrect source url: %v", err)
		}
		sources[i] = dbUrl
		uids[i] = uint32(i)
	}
	if err := mysqld.RestoreFromMultiSnapshot(dbName, keyRange, sources, uids, *concurrency, *fetchConcurrency, *insertTableConcurrency, *fetchRetryCount, *strategy); err != nil {
		relog.Fatal("multirestore failed: %v", err)
	}
}

func TestVTGateExecuteBatchKeyspaceIds(t *testing.T) {
	s := createSandbox("TestVTGateExecuteBatchKeyspaceIds")
	s.MapTestConn("-20", &sandboxConn{})
	s.MapTestConn("20-40", &sandboxConn{})
	kid10, err := key.HexKeyspaceId("10").Unhex()
	if err != nil {
		t.Errorf("want nil, got %v", err)
	}
	kid30, err := key.HexKeyspaceId("30").Unhex()
	if err != nil {
		t.Errorf("want nil, got %v", err)
	}
	q := proto.KeyspaceIdBatchQuery{
		Queries: []tproto.BoundQuery{{
			"query",
			nil,
		}, {
			"query",
			nil,
		}},
		Keyspace:    "TestVTGateExecuteBatchKeyspaceIds",
		KeyspaceIds: []key.KeyspaceId{kid10, kid30},
		TabletType:  topo.TYPE_MASTER,
	}
	qrl := new(proto.QueryResultList)
	err = RpcVTGate.ExecuteBatchKeyspaceIds(nil, &q, qrl)
	if err != nil {
		t.Errorf("want nil, got %v", err)
	}
	if len(qrl.List) != 2 {
		t.Errorf("want 2, got %v", len(qrl.List))
	}
	if qrl.List[0].RowsAffected != 2 {
		t.Errorf("want 2, got %v", qrl.List[0].RowsAffected)
	}
	if qrl.Session != nil {
		t.Errorf("want nil, got %+v\n", qrl.Session)
	}

	q.Session = new(proto.Session)
	RpcVTGate.Begin(nil, q.Session)
	err = RpcVTGate.ExecuteBatchKeyspaceIds(nil, &q, qrl)
	if len(q.Session.ShardSessions) != 2 {
		t.Errorf("want 2, got %d", len(q.Session.ShardSessions))
	}
}

func TestResolverStreamExecuteKeyspaceIds(t *testing.T) {
	kid10, err := key.HexKeyspaceId("10").Unhex()
	if err != nil {
		t.Errorf(err.Error())
	}
	kid15, err := key.HexKeyspaceId("15").Unhex()
	if err != nil {
		t.Errorf(err.Error())
	}
	kid25, err := key.HexKeyspaceId("25").Unhex()
	if err != nil {
		t.Errorf(err.Error())
	}
	createSandbox("TestResolverStreamExecuteKeyspaceIds")
	testResolverStreamGeneric(t, "TestResolverStreamExecuteKeyspaceIds", func() (*mproto.QueryResult, error) {
		res := NewResolver(new(sandboxTopo), "", "aa", retryDelay, 0, connTimeoutTotal, connTimeoutPerConn, connLife)
		qr := new(mproto.QueryResult)
		err = res.StreamExecuteKeyspaceIds(context.Background(),
			"query",
			nil,
			"TestResolverStreamExecuteKeyspaceIds",
			[]key.KeyspaceId{kid10, kid15},
			pb.TabletType_MASTER,
			func(r *mproto.QueryResult) error {
				appendResult(qr, r)
				return nil
			})
		return qr, err
	})
	testResolverStreamGeneric(t, "TestResolverStreamExecuteKeyspaceIds", func() (*mproto.QueryResult, error) {
		res := NewResolver(new(sandboxTopo), "", "aa", retryDelay, 0, connTimeoutTotal, connTimeoutPerConn, connLife)
		qr := new(mproto.QueryResult)
		err = res.StreamExecuteKeyspaceIds(context.Background(),
			"query",
			nil,
			"TestResolverStreamExecuteKeyspaceIds",
			[]key.KeyspaceId{kid10, kid15, kid25},
			pb.TabletType_MASTER,
			func(r *mproto.QueryResult) error {
				appendResult(qr, r)
				return nil
			})
		return qr, err
	})
}

func (blServer *BinlogServer) ServeBinlog(req *proto.BinlogServerRequest, sendReply proto.SendBinlogResponse) error {
	defer func() {
		if x := recover(); x != nil {
			//Send the error to the client.
			_, ok := x.(*BinlogServerError)
			if !ok {
				log.Errorf("Uncaught panic at top-most level: '%v'", x)
				//panic(x)
			}
			sendError(sendReply, req.StartPosition.String(), x.(error), nil)
		}
	}()

	log.Infof("received req: %v kr start %v end %v", req.StartPosition.String(), req.KeyspaceStart, req.KeyspaceEnd)
	if !blServer.isServiceEnabled() {
		panic(newBinlogServerError("Binlog Server is disabled"))
	}
	if !isRequestValid(req) {
		panic(newBinlogServerError("Invalid request, cannot serve the stream"))
	}

	binlogPrefix := blServer.mycnf.BinLogPath
	logsDir := path.Dir(binlogPrefix)
	if !IsMasterPositionValid(&req.StartPosition) {
		panic(newBinlogServerError(fmt.Sprintf("Invalid start position %v, cannot serve the stream, cannot locate start position", req.StartPosition)))
	}

	startKey, err := key.HexKeyspaceId(req.KeyspaceStart).Unhex()
	if err != nil {
		panic(newBinlogServerError(fmt.Sprintf("Unhex on key '%v' failed", req.KeyspaceStart)))
	}
	endKey, err := key.HexKeyspaceId(req.KeyspaceEnd).Unhex()
	if err != nil {
		panic(newBinlogServerError(fmt.Sprintf("Unhex on key '%v' failed", req.KeyspaceEnd)))
	}
	keyRange := &key.KeyRange{Start: startKey, End: endKey}

	blp := newBls(&req.StartPosition, blServer, keyRange)
	blp.binlogPrefix = binlogPrefix

	log.Infof("blp.binlogPrefix %v logsDir %v", blp.binlogPrefix, logsDir)
	blp.streamBinlog(sendReply, blServer.interrupted)
	return nil
}

// ReadRecord returns a keyspaceId and a line from which it was
// extracted, with the keyspaceId stripped.
func (r KeyspaceCSVReader) ReadRecord() (keyspaceId key.KeyspaceId, line []byte, err error) {
	k, err := r.reader.ReadString(r.delim)
	if err != nil {
		return key.MinKey, nil, err
	}
	if r.numberColumn {
		// the line starts with:
		// NNNN,
		// so remove the comma
		kid, err := strconv.ParseUint(k[:len(k)-1], 10, 64)
		if err != nil {
			return key.MinKey, nil, err
		}
		keyspaceId = key.Uint64Key(kid).KeyspaceId()
	} else {
		// the line starts with:
		// "HHHH",
		// so remove the quotes and comma
		keyspaceId, err = key.HexKeyspaceId(k[1 : len(k)-2]).Unhex()
		if err != nil {
			return key.MinKey, nil, err
		}
	}

	defer r.buf.Reset()

	escaped := false
	inQuote := false
	for {
		b, err := r.reader.ReadByte()
		if err != nil {
			// Assumption: the csv file ends with a
			// newline. Otherwise io.EOF should be treated
			// separately.
			return key.MinKey, nil, err
		}

		r.buf.WriteByte(b)

		if escaped {
			escaped = false
			continue
		}
		switch b {
		case '\\':
			escaped = true
		case '"':
			inQuote = !inQuote
		case '\n':
			if !inQuote {
				return keyspaceId, r.buf.Bytes(), nil
			}
		}
	}
}

func TestResolverStreamExecuteKeyRanges(t *testing.T) {
	kid10, err := key.HexKeyspaceId("10").Unhex()
	if err != nil {
		t.Errorf(err.Error())
	}
	kid15, err := key.HexKeyspaceId("15").Unhex()
	if err != nil {
		t.Errorf(err.Error())
	}
	kid25, err := key.HexKeyspaceId("25").Unhex()
	if err != nil {
		t.Errorf(err.Error())
	}
	query := &proto.KeyRangeQuery{
		Sql:        "query",
		Keyspace:   "TestResolverStreamExecuteKeyRanges",
		KeyRanges:  []key.KeyRange{key.KeyRange{Start: kid10, End: kid15}},
		TabletType: topo.TYPE_MASTER,
	}
	createSandbox("TestResolverStreamExecuteKeyRanges")
	// streaming a single shard
	testResolverStreamGeneric(t, "TestResolverStreamExecuteKeyRanges", func() (*mproto.QueryResult, error) {
		res := NewResolver(new(sandboxTopo), "aa", 1*time.Millisecond, 0, 1*time.Millisecond)
		qr := new(mproto.QueryResult)
		err = res.StreamExecuteKeyRanges(nil, query, func(r *mproto.QueryResult) error {
			appendResult(qr, r)
			return nil
		})
		return qr, err
	})
	// streaming multiple shards
	testResolverStreamGeneric(t, "TestResolverStreamExecuteKeyRanges", func() (*mproto.QueryResult, error) {
		query.KeyRanges = []key.KeyRange{key.KeyRange{Start: kid10, End: kid25}}
		res := NewResolver(new(sandboxTopo), "aa", 1*time.Millisecond, 0, 1*time.Millisecond)
		qr := new(mproto.QueryResult)
		err = res.StreamExecuteKeyRanges(nil, query, func(r *mproto.QueryResult) error {
			appendResult(qr, r)
			return nil
		})
		return qr, err
	})
}