func (c *RaftCluster) HandleAskBatchSplit(request *pdpb.AskBatchSplitRequest) (*pdpb.AskBatchSplitResponse, error) {
	reqRegion := request.GetRegion()
	splitCount := request.GetSplitCount()
	err := c.ValidRequestRegion(reqRegion)
	if err != nil {
		return nil, err
	}
	splitIDs := make([]*pdpb.SplitID, 0, splitCount)
	recordRegions := make([]uint64, 0, splitCount+1)
	for i := 0; i < int(splitCount); i++ {
		newRegionID, err := c.id.Alloc()
		if err != nil {
			return nil, errs.ErrSchedulerNotFound.FastGenByArgs()
		}
		peerIDs := make([]uint64, len(request.Region.Peers))
		for i := 0; i < len(peerIDs); i++ {
			if peerIDs[i], err = c.id.Alloc(); err != nil {
				return nil, err
			}
		}
		recordRegions = append(recordRegions, newRegionID)
		splitIDs = append(splitIDs, &pdpb.SplitID{
			NewRegionId: newRegionID,
			NewPeerIds:  peerIDs,
		})
		log.Info("alloc ids for region split", zap.Uint64("region-id", newRegionID), zap.Uint64s("peer-ids", peerIDs))
	}
	recordRegions = append(recordRegions, reqRegion.GetId())
	if c.IsFeatureSupported(versioninfo.RegionMerge) {
		c.GetMergeChecker().RecordRegionSplit(recordRegions)
	}
	c.AddSuspectRegions(recordRegions...)
	resp := &pdpb.AskBatchSplitResponse{Ids: splitIDs}
	return resp, nil
}
HandleAskBatchSplit handles the batch split request.
package cluster
import (
	"bytes"
	"github.com/gogo/protobuf/proto"
	"github.com/pingcap/errors"
	"github.com/pingcap/kvproto/pkg/metapb"
	"github.com/pingcap/kvproto/pkg/pdpb"
	"github.com/pingcap/log"
	"github.com/tikv/pd/pkg/errs"
	"github.com/tikv/pd/pkg/logutil"
	"github.com/tikv/pd/server/core"
	"github.com/tikv/pd/server/schedule"
	"github.com/tikv/pd/server/versioninfo"
	"go.uber.org/zap"
)
func (c *RaftCluster) HandleRegionHeartbeat(region *core.RegionInfo) error {
	if err := c.processRegionHeartbeat(region); err != nil {
		return err
	}
	c.RLock()
	co := c.coordinator
	c.RUnlock()
	co.opController.Dispatch(region, schedule.DispatchFromHeartBeat)
	return nil
}
func (c *RaftCluster) HandleAskSplit(request *pdpb.AskSplitRequest) (*pdpb.AskSplitResponse, error) {
	reqRegion := request.GetRegion()
	err := c.ValidRequestRegion(reqRegion)
	if err != nil {
		return nil, err
	}
	newRegionID, err := c.id.Alloc()
	if err != nil {
		return nil, err
	}
	peerIDs := make([]uint64, len(request.Region.Peers))
	for i := 0; i < len(peerIDs); i++ {
		if peerIDs[i], err = c.id.Alloc(); err != nil {
			return nil, err
		}
	}
	if c.IsFeatureSupported(versioninfo.RegionMerge) {
		c.GetMergeChecker().RecordRegionSplit([]uint64{reqRegion.GetId(), newRegionID})
	}
	split := &pdpb.AskSplitResponse{
		NewRegionId: newRegionID,
		NewPeerIds:  peerIDs,
	}
	log.Info("alloc ids for region split", zap.Uint64("region-id", newRegionID), zap.Uint64s("peer-ids", peerIDs))
	return split, nil
}
func (c *RaftCluster) ValidRequestRegion(reqRegion *metapb.Region) error {
	startKey := reqRegion.GetStartKey()
	region := c.GetRegionByKey(startKey)
	if region == nil {
		return errors.Errorf("region not found, request region: %v", logutil.RedactStringer(core.RegionToHexMeta(reqRegion)))
	}
	reqRegionEpoch := reqRegion.GetRegionEpoch()
	regionEpoch := region.GetMeta().GetRegionEpoch()
	if reqRegionEpoch.GetVersion() < regionEpoch.GetVersion() ||
		reqRegionEpoch.GetConfVer() < regionEpoch.GetConfVer() {
		return errors.Errorf("invalid region epoch, request: %v, current: %v", reqRegionEpoch, regionEpoch)
	}
	return nil
}