From 3c15199bdc344285bbdd5da2ee3f7b35ebbbb888 Mon Sep 17 00:00:00 2001 From: abhinavdangeti Date: Thu, 27 Aug 2020 17:04:00 -0600 Subject: [PATCH] MB-41144: Notify manager only once on feed closure + On feed closure due to any error seen while processing DCP messages, notify manager only once for the feed. Change-Id: I19d9060dd40c00f141ec008fdc23869c86ae4233 Reviewed-on: http://review.couchbase.org/c/cbgt/+/135209 Well-Formed: Build Bot Reviewed-by: Sreekanth Sivasankaran Tested-by: Abhinav Dangeti --- feed_dcp_gocbcore.go | 31 ++++++++++++++++++++++++------- 1 file changed, 24 insertions(+), 7 deletions(-) diff --git a/feed_dcp_gocbcore.go b/feed_dcp_gocbcore.go index 44458752..52f6a858 100644 --- a/feed_dcp_gocbcore.go +++ b/feed_dcp_gocbcore.go @@ -602,10 +602,28 @@ func (f *GocbcoreDCPFeed) Start() error { } func (f *GocbcoreDCPFeed) Close() error { + if f.close() { + log.Printf("feed_dcp_gocbcore: Close, name: %s", f.Name()) + } + return nil +} + +func (f *GocbcoreDCPFeed) notifyMgrOnClose() { + if f.close() { + log.Printf("feed_dcp_gocbcore: Close, name: %s, notify manager", + f.Name()) + + if f.mgr != nil { + f.mgr.Kick("gocbcore-feed") + } + } +} + +func (f *GocbcoreDCPFeed) close() bool { f.m.Lock() if f.closed { f.m.Unlock() - return nil + return false } f.closed = true go f.agent.Close() @@ -619,8 +637,7 @@ func (f *GocbcoreDCPFeed) Close() error { close(f.closeCh) f.wait() - log.Printf("feed_dcp_gocbcore: Close, name: %s", f.Name()) - return nil + return true } func (f *GocbcoreDCPFeed) Dests() map[string]Dest { @@ -777,10 +794,10 @@ func (f *GocbcoreDCPFeed) onError(notifyMgr bool, err error) error { " bucketName: %s, bucketUUID: %s, err: %v", f.Name(), f.bucketName, f.bucketUUID, err) - f.Close() - - if notifyMgr && f.mgr != nil { - f.mgr.Kick("gocbcore-feed") + if notifyMgr { + f.notifyMgrOnClose() + } else { + f.Close() } return err