MB-39662 add stream reset message to mutation manager

Change-Id: I193f623af59db2da3807fdbde498a4e70ca7a410
couchbase · Oct 13, 2020 · a963e87 · a963e87
1 parent b322f7a
commit a963e87
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 2 deletions.
diff --git a/secondary/indexer/mutation_manager.go b/secondary/indexer/mutation_manager.go
@@ -359,7 +359,8 @@ func (m *mutationMgr) handleWorkerMessage(cmd Message) {
 		STREAM_READER_CONN_ERROR,
 		STREAM_READER_HWT,
 		STREAM_READER_SYSTEM_EVENT,
-		STREAM_READER_OSO_SNAPSHOT_MARKER:
+		STREAM_READER_OSO_SNAPSHOT_MARKER,
+		RESET_STREAM:
 		//send message to supervisor to take decision
 		logging.Tracef("MutationMgr::handleWorkerMessage Received %v from worker", cmd)
 		m.supvRespch <- cmd

diff --git a/secondary/indexer/stream_reader.go b/secondary/indexer/stream_reader.go
@@ -1110,7 +1110,7 @@ func (w *streamWorker) updateOSOMarkerInFilter(meta *MutationMeta, eventType byt
 		}
 
 		resetStream := func() {
-			logging.Infof("MutationStreamReader::updateOSOMarkerInFilter %v %v.",
+			logging.Infof("MutationStreamReader::updateOSOMarkerInFilter %v %v."+
 				" Resetting Stream.", w.streamId, meta.keyspaceId)
 			w.keyspaceIdOSOException[meta.keyspaceId] = true
 			w.reader.supvRespch <- &MsgStreamUpdate{