Merge pull request #13946 from ahrtr/move_cindex_on_apply_fail_353

[3.5] Update consitent_index when applying fails

server/etcdserver/server.go

@@ -2167,18 +2167,19 @@ func (s *EtcdServer) apply(
 func (s *EtcdServer) applyEntryNormal(e *raftpb.Entry) {
 	shouldApplyV3 := membership.ApplyV2storeOnly
 	applyV3Performed := false
-	defer func() {
-		// The txPostLock callback will not get called in this case,
-		// so we should set the consistent index directly.
-		if s.consistIndex != nil && !applyV3Performed && membership.ApplyBoth == shouldApplyV3 {
-			s.consistIndex.SetConsistentIndex(e.Index, e.Term)
-		}
-	}()
+	var ar *applyResult
 	index := s.consistIndex.ConsistentIndex()
 	if e.Index > index {
 		// set the consistent index of current executing entry
 		s.consistIndex.SetConsistentApplyingIndex(e.Index, e.Term)
 		shouldApplyV3 = membership.ApplyBoth
+		defer func() {
+			// The txPostLockInsideApplyHook will not get called in some cases,
+			// in which we should move the consistent index forward directly.
+			if !applyV3Performed || (ar != nil && ar.err != nil) {
+				s.consistIndex.SetConsistentIndex(e.Index, e.Term)
+			}
+		}()
 	}
 	s.lg.Debug("apply entry normal",
 		zap.Uint64("consistent-index", index),
@@ -2220,7 +2221,6 @@ func (s *EtcdServer) applyEntryNormal(e *raftpb.Entry) {
 		id = raftReq.Header.ID
 	}
 
-	var ar *applyResult
 	needResult := s.w.IsRegistered(id)
 	if needResult || !noSideEffect(&raftReq) {
 		if !needResult && raftReq.Txn != nil {

tests/integration/v3_auth_test.go

@@ -46,6 +46,33 @@ func TestV3AuthEmptyUserGet(t *testing.T) {
 	}
 }
 
+// TestV3AuthEmptyUserPut ensures that a put with an empty user will return an empty user error,
+// and the consistent_index should be moved forward even the apply-->Put fails.
+func TestV3AuthEmptyUserPut(t *testing.T) {
+	BeforeTest(t)
+	clus := NewClusterV3(t, &ClusterConfig{
+		Size:          1,
+		SnapshotCount: 3,
+	})
+	defer clus.Terminate(t)
+
+	ctx, cancel := context.WithTimeout(context.TODO(), 30*time.Second)
+	defer cancel()
+
+	api := toGRPC(clus.Client(0))
+	authSetupRoot(t, api.Auth)
+
+	// The SnapshotCount is 3, so there must be at least 3 new snapshot files being created.
+	// The VERIFY logic will check whether the consistent_index >= last snapshot index on
+	// cluster terminating.
+	for i := 0; i < 10; i++ {
+		_, err := api.KV.Put(ctx, &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")})
+		if !eqErrGRPC(err, rpctypes.ErrUserEmpty) {
+			t.Fatalf("got %v, expected %v", err, rpctypes.ErrUserEmpty)
+		}
+	}
+}
+
 // TestV3AuthTokenWithDisable tests that auth won't crash if
 // given a valid token when authentication is disabled
 func TestV3AuthTokenWithDisable(t *testing.T) {