Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #14995 from ahrtr/revert_14322_20221215

Browse Source 
clientv3: revert the client side change in 14547
This commit is contained in:
Hitoshi Mitake 2022-12-15 09:39:35 +09:00 committed by GitHub
commit f1842b6ecf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 0 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
client/v3/watch.go
View File

@ -18,7 +18,6 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"strings"
"sync" "sync"
"time" "time"
@ -589,26 +588,6 @@ func (w *watchGrpcStream) run() {
switch { switch {
case pbresp.Created: case pbresp.Created:
cancelReasonError := v3rpc.Error(errors.New(pbresp.CancelReason))
if shouldRetryWatch(cancelReasonError) {
var newErr error
if wc, newErr = w.newWatchClient(); newErr != nil {
w.lg.Error("failed to create a new watch client", zap.Error(newErr))
return
}
if len(w.resuming) != 0 {
if ws := w.resuming[0]; ws != nil {
if err := wc.Send(ws.initReq.toPB()); err != nil {
w.lg.Debug("error when sending request", zap.Error(err))
}
}
}
cur = nil
continue
}
// response to head of queue creation // response to head of queue creation
if len(w.resuming) != 0 { if len(w.resuming) != 0 {
if ws := w.resuming[0]; ws != nil { if ws := w.resuming[0]; ws != nil {
@ -718,11 +697,6 @@ func (w *watchGrpcStream) run() {
} }
} }
func shouldRetryWatch(cancelReasonError error) bool {
return (strings.Compare(cancelReasonError.Error(), v3rpc.ErrGRPCInvalidAuthToken.Error()) == 0) ||
(strings.Compare(cancelReasonError.Error(), v3rpc.ErrGRPCAuthOldRevision.Error()) == 0)
}
// nextResume chooses the next resuming to register with the grpc stream. Abandoned // nextResume chooses the next resuming to register with the grpc stream. Abandoned
// streams are marked as nil in the queue since the head must wait for its inflight registration. // streams are marked as nil in the queue since the head must wait for its inflight registration.
func (w *watchGrpcStream) nextResume() *watcherStream { func (w *watchGrpcStream) nextResume() *watcherStream {

33
tests/integration/v3_auth_test.go
View File

@ -499,39 +499,6 @@ func TestV3AuthRestartMember(t *testing.T) {
testutil.AssertNil(t, err) testutil.AssertNil(t, err)
} }
func TestV3AuthWatchAndTokenExpire(t *testing.T) {
BeforeTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 1, AuthTokenTTL: 3})
defer clus.Terminate(t)
ctx, cancel := context.WithTimeout(context.TODO(), 10*time.Second)
defer cancel()
authSetupRoot(t, toGRPC(clus.Client(0)).Auth)
c, cerr := NewClient(t, clientv3.Config{Endpoints: clus.Client(0).Endpoints(), Username: "root", Password: "123"})
if cerr != nil {
t.Fatal(cerr)
}
defer c.Close()
_, err := c.Put(ctx, "key", "val")
if err != nil {
t.Fatalf("Unexpected error from Put: %v", err)
}
// The first watch gets a valid auth token through watcher.newWatcherGrpcStream()
// We should discard the first one by waiting TTL after the first watch.
wChan := c.Watch(ctx, "key", clientv3.WithRev(1))
watchResponse := <-wChan
time.Sleep(5 * time.Second)
wChan = c.Watch(ctx, "key", clientv3.WithRev(1))
watchResponse = <-wChan
testutil.AssertNil(t, watchResponse.Err())
}
func TestV3AuthWatchErrorAndWatchId0(t *testing.T) { func TestV3AuthWatchErrorAndWatchId0(t *testing.T) {
BeforeTest(t) BeforeTest(t)
clus := NewClusterV3(t, &ClusterConfig{Size: 3}) clus := NewClusterV3(t, &ClusterConfig{Size: 3})