Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

gohbase: fix lease expired on region #104

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

gohbase: fix lease expired on region #104

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 27 additions & 19 deletions scanner.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,30 +35,32 @@ type scanner struct {
closed bool
}

func (s *scanner) closeRegionScanner(table, startRow []byte, scannerID uint64) {
// if we are closing in the middle of scanning a region,
// send a close scanner request
// TODO: add a deadline
rpc, err := hrpc.NewScanRange(context.Background(),
table, startRow, nil,
hrpc.ScannerID(scannerID),
hrpc.CloseScanner(),
hrpc.NumberOfRows(0))
if err != nil {
panic(fmt.Sprintf("should not happen: %s", err))
}

// If the request fails, the scanner lease will be expired
// and it will be closed automatically by hbase.
// No need to bother clients about that.
s.SendRPC(rpc)
}

func (s *scanner) Close() error {
if s.closed {
return errors.New("scanner has already been closed")
}
s.closed = true
if s.scannerID != noScannerID {
go func() {
// if we are closing in the middle of scanning a region,
// send a close scanner request
// TODO: add a deadline
rpc, err := hrpc.NewScanRange(context.Background(),
s.rpc.Table(), s.startRow, nil,
hrpc.ScannerID(s.scannerID),
hrpc.CloseScanner(),
hrpc.NumberOfRows(0))
if err != nil {
panic(fmt.Sprintf("should not happen: %s", err))
}

// If the request fails, the scanner lease will be expired
// and it will be closed automatically by hbase.
// No need to bother clients about that.
s.SendRPC(rpc)
}()
if s.scannerID != noScannerID && !s.rpc.IsClosing() {
go s.closeRegionScanner(s.rpc.Table(), s.startRow, s.scannerID)
}
return nil
}
Expand Down Expand Up @@ -241,12 +243,18 @@ func (s *scanner) request() (*pb.ScanResponse, hrpc.RegionInfo, error) {

// update updates the scanner for the next scan request
func (s *scanner) update(resp *pb.ScanResponse, region hrpc.RegionInfo) {
if s.scannerID == noScannerID && resp.ScannerId != nil {
s.scannerID = resp.GetScannerId()
}
if resp.GetMoreResultsInRegion() {
if resp.ScannerId != nil {
s.scannerID = resp.GetScannerId()
}
} else {
// we are done with this region, prepare scan for next region
if !s.rpc.IsClosing() {
go s.closeRegionScanner(s.rpc.Table(), s.startRow, s.scannerID)
}
s.scannerID = noScannerID

// Normal Scan
Expand Down