MB-53183 Do not check snap alignment for INIT_STREAM

BP to 7.1.2 branch

As indexer can persist non-snapshot aligned timestamps for
INIT_STREAM. Also, for initial index build in MAINT_STREAM,
do not assert for non-snap aligned timestamps

Change-Id: Ib6f3e227898b38589f45bbcb96c2d974baee5cb4

secondary/indexer/storage_manager.go

@@ -256,13 +256,6 @@ func (s *storageMgr) handleCreateSnapshot(cmd Message) {
 	snapType := tsVbuuid.GetSnapType()
 	tsVbuuid.Crc64 = common.HashVbuuid(tsVbuuid.Vbuuids)
 
-	if (snapType == common.DISK_SNAP ||
-		snapType == common.FORCE_COMMIT ||
-		snapType == common.FORCE_COMMIT_MERGE) && (tsVbuuid.CheckSnapAligned() == false) {
-		logging.Fatalf("StorageMgr::handleCreateSnapshot Disk commit timestamp is not snapshot aligned. "+
-			"Stream: %v, KeyspaceId: %v, tsVbuuid: %v", streamId, keyspaceId, tsVbuuid)
-	}
-
 	//if snapType is FORCE_COMMIT_MERGE, sync response
 	//will be sent after snasphot creation
 	if snapType != common.FORCE_COMMIT_MERGE {
@@ -315,6 +308,8 @@ func (s *storageMgr) handleCreateSnapshot(cmd Message) {
 
 	s.muSnap.Unlock()
 
+	s.assertOnNonAlignedDiskCommit(streamId, keyspaceId, tsVbuuid_copy)
+
 	if snapType == common.FORCE_COMMIT_MERGE {
 		//response is sent on supvCmdch in case of FORCE_COMMIT_MERGE
 		s.createSnapshotWorker(streamId, keyspaceId, tsVbuuid_copy, indexSnapMap,
@@ -2240,3 +2235,23 @@ func (s *storageMgr) getNumSnapshotWorkers() int {
 	}
 	return numSnapshotWorkers
 }
+
+func (s *storageMgr) assertOnNonAlignedDiskCommit(streamId common.StreamId,
+	keyspaceId string, tsVbuuid *common.TsVbuuid) {
+
+	snapType := tsVbuuid.GetSnapType()
+	// For INIT_STREAM, disk snapshot need not be snap aligned
+	// Hence, the assertion is only for MAINT_STREAM
+	// From 7.x all initial index builds happen in INIT_STREAM. So,
+	// there is no need to check for index state from 7.0 as we are
+	// doing this assertion check only for MAINT_STREAM
+	if (streamId == common.MAINT_STREAM) &&
+		(snapType == common.DISK_SNAP ||
+			snapType == common.FORCE_COMMIT ||
+			snapType == common.FORCE_COMMIT_MERGE) && (tsVbuuid.CheckSnapAligned() == false) {
+
+		logging.Fatalf("StorageMgr::handleCreateSnapshot Disk commit timestamp is not snapshot aligned. "+
+			"Stream: %v, KeyspaceId: %v, tsVbuuid: %v", streamId, keyspaceId, tsVbuuid)
+
+	}
+}