func TestSendRPCToRegionClientDownDelayed(t *testing.T) {
	ctrl := test.NewController(t)
	defer ctrl.Finish()
	c := newMockClient(nil)
	origlReg := region.NewInfo(
		0, nil, []byte("test1"),
		[]byte("test1,,1234567890042.56f833d5569a27c7a43fbf547b4924a4."),
		nil, nil)
	c.regions.put(origlReg)
	rc := mockRegion.NewMockRegionClient(ctrl)
	rc.EXPECT().String().Return("mock region client").AnyTimes()
	c.clients.put("regionserver:0", origlReg, func() hrpc.RegionClient {
		return rc
	})
	origlReg.SetClient(rc)
	mockCall := mock.NewMockCall(ctrl)
	mockCall.EXPECT().SetRegion(origlReg).Times(1)
	mockCall.EXPECT().Context().Return(context.Background())
	result := make(chan hrpc.RPCResult, 1)
	mockCall.EXPECT().ResultChan().Return(result).Times(1)
	rc2 := mockRegion.NewMockRegionClient(ctrl)
	rc2.EXPECT().String().Return("mock region client").AnyTimes()
	rc.EXPECT().QueueRPC(mockCall).Times(1).Do(func(rpc hrpc.Call) {
		c.clients.clientDown(rc)
		c.regions.put(origlReg)
		c.clients.put("regionserver:0", origlReg, func() hrpc.RegionClient {
			return rc2
		})
		origlReg.SetClient(rc2)
		result <- hrpc.RPCResult{Error: region.ServerError{}}
	})
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		_, err := c.sendRPCToRegion(mockCall, origlReg)
		switch err.(type) {
		case region.ServerError, region.NotServingRegionError:
		default:
			t.Errorf("Got unexpected error: %v", err)
		}
		wg.Done()
	}()
	wg.Wait()
	if len(c.clients.regions) != 1 {
		t.Errorf("There are %d cached clients", len(c.clients.regions))
	}
	_, ok := c.clients.regions[rc2]
	if !ok {
		t.Error("rc2 is not in clients cache")
	}
}
TestSendRPCToRegionClientDownDelayed check that the we don't replace a newly
looked up client (by other goroutine) in clients cache if we are too slow
at finding out that our old client died.
package gohbase
import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"reflect"
	"strconv"
	"sync"
	"testing"
	"time"
	"github.com/golang/mock/gomock"
	"github.com/tsuna/gohbase/compression"
	"github.com/tsuna/gohbase/hrpc"
	"github.com/tsuna/gohbase/pb"
	"github.com/tsuna/gohbase/region"
	"github.com/tsuna/gohbase/test"
	"github.com/tsuna/gohbase/test/mock"
	mockRegion "github.com/tsuna/gohbase/test/mock/region"
	mockZk "github.com/tsuna/gohbase/test/mock/zk"
	"github.com/tsuna/gohbase/zk"
	"google.golang.org/protobuf/proto"
	"modernc.org/b"
)
func newRegionClientFn(addr string) func() hrpc.RegionClient {
	return func() hrpc.RegionClient {
		return newMockRegionClient(addr, region.RegionClient,
			0, 0, "root", region.DefaultReadTimeout, nil)
	}
}
func newMockClient(zkClient zk.Client) *client {
	return &client{
		clientType: region.RegionClient,
		regions:    keyRegionCache{regions: b.TreeNew(region.CompareGeneric)},
		clients: clientRegionCache{
			regions: make(map[hrpc.RegionClient]map[hrpc.RegionInfo]struct{}),
		},
		rpcQueueSize:  defaultRPCQueueSize,
		flushInterval: defaultFlushInterval,
		metaRegionInfo: region.NewInfo(0, []byte("hbase"), []byte("meta"),
			[]byte("hbase:meta,,1"), nil, nil),
		zkTimeout:           defaultZkTimeout,
		zkClient:            zkClient,
		regionLookupTimeout: region.DefaultLookupTimeout,
		regionReadTimeout:   region.DefaultReadTimeout,
		newRegionClientFn:   newMockRegionClient,
	}
}
func TestSendRPCSanity(t *testing.T) {
	ctrl := test.NewController(t)
	defer ctrl.Finish()
	zkClient := mockZk.NewMockClient(ctrl)
	zkClient.EXPECT().LocateResource(zk.Meta).Return("regionserver:1", nil).MinTimes(1)
	c := newMockClient(zkClient)
	mockCall := mock.NewMockCall(ctrl)
	mockCall.EXPECT().Context().Return(context.Background()).AnyTimes()
	mockCall.EXPECT().SetContext(gomock.Any()).AnyTimes()
	mockCall.EXPECT().Description().AnyTimes()
	mockCall.EXPECT().Table().Return([]byte("test")).AnyTimes()
	mockCall.EXPECT().Key().Return([]byte("theKey")).AnyTimes()
	mockCall.EXPECT().SetRegion(gomock.Any()).AnyTimes()
	result := make(chan hrpc.RPCResult, 1)
	expMsg := &pb.ScanResponse{}
	result <- hrpc.RPCResult{Msg: expMsg}
	mockCall.EXPECT().ResultChan().Return(result).Times(1)
	msg, err := c.SendRPC(mockCall)
	if err != nil {
		t.Fatal(err)
	}
	if !proto.Equal(expMsg, msg) {
		t.Errorf("expected %v, got %v", expMsg, msg)
	}
	if len(c.clients.regions) != 2 {
		t.Errorf("Expected 2 clients in cache, got %d", len(c.clients.regions))
	}
	expClients := map[string]string{
		"regionserver:1": "hbase:meta,,1",
		"regionserver:2": "test,,1434573235908.56f833d5569a27c7a43fbf547b4924a4.",
	}
	for c, rs := range c.clients.regions {
		name, ok := expClients[c.Addr()]
		if !ok {
			t.Errorf("Got unexpected client %s in cache", c.Addr())
			continue
		}
		if len(rs) != 1 {
			t.Errorf("Expected to have only 1 region in cache for client %s", c.Addr())
			continue
		}
		for r := range rs {
			if string(r.Name()) != name {
				t.Errorf("Unexpected name of region %q for client %s, expected %q",
					r.Name(), c.Addr(), name)
			}
			rc := r.Client()
			if c != rc {
				t.Errorf("Invalid bidirectional mapping: forward=%s, backward=%s",
					c.Addr(), rc.Addr())
			}
		}
	}
	if c.regions.regions.Len() != 1 {
		t.Errorf("Expected 1 regions in cache, got %d", c.regions.regions.Len())
	}
}
func TestReestablishRegionSplit(t *testing.T) {
	ctrl := test.NewController(t)
	defer ctrl.Finish()
	c := newMockClient(mockZk.NewMockClient(ctrl))
	origlReg := region.NewInfo(
		0,
		nil,
		[]byte("test1"),
		[]byte("test1,,1434573235908.56f833d5569a27c7a43fbf547b4924a4."),
		nil,
		nil,
	)
	origlReg.MarkUnavailable()
	rc1 := c.clients.put("regionserver:1", origlReg, newRegionClientFn("regionserver:1"))
	origlReg.SetClient(rc1)
	c.regions.put(origlReg)
	rc2 := c.clients.put("regionserver:1", c.metaRegionInfo, newRegionClientFn("regionserver:1"))
	if rc1 != rc2 {
		t.Fatal("expected to get the same region client")
	}
	c.metaRegionInfo.SetClient(rc2)
	c.reestablishRegion(origlReg)
	if len(c.clients.regions) != 1 {
		t.Errorf("Expected 1 client in cache, got %d", len(c.clients.regions))
	}
	expRegs := map[string]struct{}{
		"hbase:meta,,1": struct{}{},
		"test1,,1480547738107.825c5c7e480c76b73d6d2bad5d3f7bb8.": struct{}{},
	}
	for rc, rs := range c.clients.regions {
		if rc.Addr() != "regionserver:1" {
			t.Errorf("Got unexpected client %s in cache", rc.Addr())
			break
		}
		gotRegs := map[string]struct{}{}
		for r := range rs {
			gotRegs[string(r.Name())] = struct{}{}
			if r.Client() != rc1 {
				t.Errorf("Invalid bidirectional mapping: forward=%s, backward=%s",
					r.Client().Addr(), rc1.Addr())
			}
		}
		if !reflect.DeepEqual(expRegs, gotRegs) {
			t.Errorf("expected %v, got %v", expRegs, gotRegs)
		}
		if rc != rc1 {
			t.Errorf("Invalid bidirectional mapping: forward=%s, backward=%s",
				rc.Addr(), rc1.Addr())
		}
	}
	if c.regions.regions.Len() != 1 {
		t.Errorf("Expected 1 regions in cache, got %d", c.regions.regions.Len())
	}
	newRegIntf, ok := c.regions.regions.Get(
		[]byte("test1,,1480547738107.825c5c7e480c76b73d6d2bad5d3f7bb8."))
	if !ok {
		t.Error("Expected region is not in the cache")
	}
	newReg, ok := newRegIntf.(hrpc.RegionInfo)
	if !ok {
		t.Error("Expected hrpc.RegionInfo")
	}
	if newReg.IsUnavailable() {
		t.Error("Expected new region to be available")
	}
	if origlReg.IsUnavailable() {
		t.Error("Expected original region to be available")
	}
	if origlReg.Client() != nil {
		t.Error("Expected original region to have no client")
	}
}
func TestReestablishRegionNSRE(t *testing.T) {
	c := newMockClient(nil)
	origlReg := region.NewInfo(0, nil, []byte("nsre"),
		[]byte("nsre,,1434573235908.56f833d5569a27c7a43fbf547b4924a4."), nil, nil)
	rc1 := c.clients.put("regionserver:1", c.metaRegionInfo, newRegionClientFn("regionserver:1"))
	rc2 := c.clients.put("regionserver:1", origlReg, newRegionClientFn("regionserver:1"))
	if rc1 != rc2 {
		t.Fatal("expected region client to be the same")
	}
	c.metaRegionInfo.SetClient(rc1)
	origlReg.SetClient(rc1)
	origlReg.MarkUnavailable()
	c.regions.put(origlReg)
	c.reestablishRegion(origlReg)
	if len(c.clients.regions) != 1 {
		t.Errorf("Expected 1 client in cache, got %d", len(c.clients.regions))
	}
	if c.regions.regions.Len() != 1 {
		t.Errorf("Expected 1 regions in cache, got %d", c.regions.regions.Len())
	}
	_, ok := c.regions.regions.Get(
		[]byte("nsre,,1434573235908.56f833d5569a27c7a43fbf547b4924a4."))
	if !ok {
		t.Error("Expected region is not in the cache")
	}
	if origlReg.IsUnavailable() {
		t.Error("Expected original region to be available")
	}
	if origlReg.Client() != rc1 {
		t.Error("Expected original region the same client")
	}
}
func TestEstablishRegionDialFail(t *testing.T) {
	ctrl := test.NewController(t)
	defer ctrl.Finish()
	c := newMockClient(nil)
	rcFailDial := mockRegion.NewMockRegionClient(ctrl)
	rcFailDial.EXPECT().Dial(gomock.Any()).Return(errors.New("ooops")).AnyTimes()
	rcFailDial.EXPECT().Addr().Return("regionserver:1").AnyTimes()
	rcFailDial.EXPECT().String().Return("regionserver:1").AnyTimes()
	rcDialCancel := mockRegion.NewMockRegionClient(ctrl)
	rcDialCancel.EXPECT().Dial(gomock.Any()).Return(context.Canceled)
	rcDialCancel.EXPECT().Addr().Return("regionserver:1").AnyTimes()
	rcDialCancel.EXPECT().String().Return("reginserver:1").AnyTimes()
	newRegionClientFnCallCount := 0
	c.newRegionClientFn = func(_ string, _ region.ClientType, _ int, _ time.Duration,
		_ string, _ time.Duration, _ compression.Codec) hrpc.RegionClient {
		var rc hrpc.RegionClient
		if newRegionClientFnCallCount == 0 {
			rc = rcFailDial
		} else {
			rc = rcDialCancel
		}
		newRegionClientFnCallCount++
		return rc
	}
	reg := region.NewInfo(
		0, nil, []byte("test1"), []byte("test1,,1434573235908.56f833d5569a27c7a43fbf547b4924a4."),
		nil, nil)
	reg.MarkUnavailable()
	rc1 := c.clients.put("regionserver:0", c.metaRegionInfo, newRegionClientFn("regionserver:0"))
	c.metaRegionInfo.SetClient(rc1)
	c.establishRegion(reg, "regionserver:1")
	if len(c.clients.regions) != 2 {
		t.Errorf("Expected 2 clients in cache, got %d", len(c.clients.regions))
	}
	if reg.IsUnavailable() {
		t.Error("Expected region to be available")
	}
}
func TestEstablishClientConcurrent(t *testing.T) {
	ctrl := test.NewController(t)
	defer ctrl.Finish()
	c := newMockClient(mockZk.NewMockClient(ctrl))
	numRegions := 1000
	regions := make([]hrpc.RegionInfo, numRegions)
	for i := range regions {
		r := region.NewInfo(
			0,
			nil,
			[]byte("test"),
			[]byte(fmt.Sprintf("test,%d,1434573235908.yoloyoloyoloyoloyoloyoloyoloyolo.", i)),
			nil, nil)
		r.MarkUnavailable()
		regions[i] = r
	}
	var wg sync.WaitGroup
	for _, r := range regions {
		r := r
		wg.Add(1)
		go func() {
			c.establishRegion(r, "regionserver:1")
			wg.Done()
		}()
	}
	wg.Wait()
	if len(c.clients.regions) != 1 {
		t.Fatalf("Expected to have only 1 client in cache, got %d", len(c.clients.regions))
	}
	for rc, rs := range c.clients.regions {
		if len(rs) != numRegions {
			t.Errorf("Expected to have %d regions, got %d", numRegions, len(rs))
		}
		for _, r := range regions {
			if r.Client() != rc {
				t.Errorf("Region %q has unexpected client %s", r.Name(), rc.Addr())
			}
			if r.IsUnavailable() {
				t.Errorf("Expected region %s to be available", r.Name())
			}
		}
	}
}
func TestEstablishServerErrorDuringProbe(t *testing.T) {
	ctrl := test.NewController(t)
	defer ctrl.Finish()
	c := newMockClient(nil)
	rc := c.clients.put("regionserver:0", c.metaRegionInfo, newRegionClientFn("regionserver:0"))
	c.metaRegionInfo.SetClient(rc)
	mockCall := mock.NewMockCall(ctrl)
	mockCall.EXPECT().Context().Return(context.Background()).AnyTimes()
	mockCall.EXPECT().SetContext(gomock.Any()).AnyTimes()
	mockCall.EXPECT().Description().AnyTimes()
	mockCall.EXPECT().Table().Return([]byte("down")).AnyTimes()
	mockCall.EXPECT().Key().Return([]byte("yolo")).AnyTimes()
	mockCall.EXPECT().SetRegion(gomock.Any()).AnyTimes()
	result := make(chan hrpc.RPCResult, 1)
	expMsg := &pb.GetResponse{}
	result <- hrpc.RPCResult{Msg: expMsg}
	mockCall.EXPECT().ResultChan().Return(result).Times(1)
	msg, err := c.SendRPC(mockCall)
	if err != nil {
		t.Errorf("Unexpected error: %s", err)
	}
	if !proto.Equal(expMsg, msg) {
		t.Errorf("expected %v, got %v", expMsg, msg)
	}
	if len(c.clients.regions) != 2 {
		t.Errorf("expected to have 2 clients in cache, got %d", len(c.clients.regions))
	}
}