MB-36095: On pause, send filter messages to all consumers on all assi…

…gned VBs

Change-Id: I38c52bca99a0422d24ba1b81fcfce7c8f111d4dc
Reviewed-on: http://review.couchbase.org/116429
Reviewed-by: <satya.nand@couchbase.com>
Tested-by: Jeelan Basha Poola <jeelan.poola@couchbase.com>

common/common.go

@@ -237,6 +237,7 @@ type EventingConsumer interface {
 	WorkerVbMapUpdate(map[string][]uint16)
 
 	SendAssignedVbs()
+	PauseConsumer()
 }
 
 type EventingSuperSup interface {

consumer/exported_functions.go

@@ -3,6 +3,7 @@ package consumer
 import (
 	"bufio"
 	"fmt"
+	"math"
 	"net"
 	"os"
 	"os/exec"
@@ -747,3 +748,13 @@ func (c *Consumer) GetInsight() *common.Insight {
 		return nil
 	}
 }
+
+func (c *Consumer) PauseConsumer() {
+	// send vb filter messages to all assigned VBs to stop processing mutations. Note the use math.MaxUint64 for seqno
+	assignedVbs, _ := c.getAssignedVbs(c.ConsumerName())
+	for _, vb := range assignedVbs {
+		c.sendVbFilterData(vb, math.MaxUint64, false)
+	}
+
+	c.WorkerVbMapUpdate(nil)
+}

consumer/process_events.go

@@ -29,7 +29,7 @@ func (c *Consumer) processDCPEvents() {
 	for {
 		if c.cppQueueSizes != nil {
 			if c.workerQueueCap < c.cppQueueSizes.AggQueueSize ||
-				c.workerQueueMemCap < (c.sentEventsSize - c.cppQueueSizes.ProcessedEventsSize) {
+				c.workerQueueMemCap < (c.sentEventsSize-c.cppQueueSizes.ProcessedEventsSize) {
 				logging.Debugf("%s [%s:%s:%d] Throttling, cpp queue sizes: %+v, event size: %d",
 					logPrefix, c.workerName, c.tcpPort, c.Pid(), c.cppQueueSizes, c.sentEventsSize)
 				time.Sleep(10 * time.Millisecond)
@@ -1313,7 +1313,9 @@ func (c *Consumer) handleStreamEnd(vBucket uint16, last_processed_seqno uint64)
 	}
 
 	c.filterVbEventsRWMutex.Lock()
-	delete(c.filterVbEvents, vBucket)
+	if _, ok := c.filterVbEvents[vBucket]; ok {
+		delete(c.filterVbEvents, vBucket)
+	}
 	c.filterVbEventsRWMutex.Unlock()
 
 	var vbBlob vbucketKVBlob

producer/producer.go

@@ -268,7 +268,7 @@ func (p *Producer) Serve() {
 			logging.Infof("%s [%s:%d] Pausing processing", logPrefix, p.appName, p.LenRunningConsumers())
 
 			for _, c := range p.getConsumers() {
-				c.WorkerVbMapUpdate(nil)
+				c.PauseConsumer()
 				c.ResetBootstrapDone()
 				c.CloseAllRunningDcpFeeds()
 			}