diff --git a/proxy/grpcproxy/watch_broadcast.go b/proxy/grpcproxy/watch_broadcast.go index d88ccfaa2..a82c842fb 100644 --- a/proxy/grpcproxy/watch_broadcast.go +++ b/proxy/grpcproxy/watch_broadcast.go @@ -54,13 +54,20 @@ func newWatchBroadcast(wp *watchProxy, w *watcher, update func(*watchBroadcast)) defer close(wb.donec) // loop because leader loss will close channel for cctx.Err() == nil { - wch := wp.cw.Watch(cctx, w.wr.key, + opts := []clientv3.OpOption{ clientv3.WithRange(w.wr.end), clientv3.WithProgressNotify(), - clientv3.WithCreatedNotify(), clientv3.WithRev(wb.nextrev), clientv3.WithPrevKV(), - ) + } + // The create notification should be the first response; + // if the watch is recreated following leader loss, it + // shouldn't post a second create response to the client. + if wb.responses == 0 { + opts = append(opts, clientv3.WithCreatedNotify()) + } + wch := wp.cw.Watch(cctx, w.wr.key, opts...) + for wr := range wch { wb.bcast(wr) update(wb)