resource_control: watch delete with prev and refine test

client/resource_group/controller/controller.go

@@ -219,7 +219,8 @@ func (c *ResourceGroupsController) Start(ctx context.Context) {
 		}
 		var watchChannel chan []*meta_storagepb.Event
 		if !c.ruConfig.isSingleGroupByKeyspace {
-			watchChannel, err = c.provider.Watch(ctx, pd.GroupSettingsPathPrefixBytes, pd.WithRev(revision), pd.WithPrefix())
+			// Use WithPrevKV() to get the previous key-value pair when get Delete Event.
+			watchChannel, err = c.provider.Watch(ctx, pd.GroupSettingsPathPrefixBytes, pd.WithRev(revision), pd.WithPrefix(), pd.WithPrevKV())
 		}
 		watchRetryTimer := time.NewTimer(watchRetryInterval)
 		if err == nil || c.ruConfig.isSingleGroupByKeyspace {
@@ -277,19 +278,29 @@ func (c *ResourceGroupsController) Start(ctx context.Context) {
 				}
 				for _, item := range resp {
 					revision = item.Kv.ModRevision
-					group := &rmpb.ResourceGroup{}
-					if err := proto.Unmarshal(item.Kv.Value, group); err != nil {
-						continue
-					}
 					switch item.Type {
 					case meta_storagepb.Event_PUT:
+						group := &rmpb.ResourceGroup{}
+						if err := proto.Unmarshal(item.Kv.Value, group); err != nil {
+							continue
+						}
 						if item, ok := c.groupsController.Load(group.Name); ok {
 							gc := item.(*groupCostController)
 							gc.modifyMeta(group)
 						}
 					case meta_storagepb.Event_DELETE:
-						if _, ok := c.groupsController.LoadAndDelete(group.Name); ok {
-							resourceGroupStatusGauge.DeleteLabelValues(group.Name)
+						if item.PrevKv != nil {
+							group := &rmpb.ResourceGroup{}
+							if err := proto.Unmarshal(item.PrevKv.Value, group); err != nil {
+								continue
+							}
+							if _, ok := c.groupsController.LoadAndDelete(group.Name); ok {
+								resourceGroupStatusGauge.DeleteLabelValues(group.Name)
+							}
+						} else {
+							// Prev-kv is compacted means there must have been a delete event before this event,
+							// which means that this is just a duplicated event, so we can just ignore it.
+							log.Info("previous key-value pair has been compacted", zap.String("previous key", string(item.Kv.Key)))
 						}
 					}
 				}