Extend support for stream message in UpdateVReplicationWorkflow

Signed-off-by: Noble Mittal <[email protected]>

go/vt/vtctl/workflow/server.go

@@ -128,7 +128,6 @@ const (
 	// Time to wait between LOCK TABLES cycles on the sources during SwitchWrites.
 	lockTablesCycleDelay = time.Duration(100 * time.Millisecond)
 
-	SqlFreezeWorkflow   = "update _vt.vreplication set message = 'FROZEN' where db_name=%s and workflow=%s"
 	SqlUnfreezeWorkflow = "update _vt.vreplication set state='Running', message='' where db_name=%s and workflow=%s"
 )
 
@@ -725,7 +724,7 @@ func (s *Server) LookupVindexExternalize(ctx context.Context, req *vtctldatapb.L
 			}
 			resp.WorkflowDeleted = true
 		} else {
-			// Stop the workflow.
+			// Freeze the workflow.
 			err = forAllShards(targetShards, func(si *topo.ShardInfo) error {
 				tabletInfo, err := s.ts.GetTablet(ctx, si.PrimaryAlias)
 				if err != nil {
@@ -734,14 +733,12 @@ func (s *Server) LookupVindexExternalize(ctx context.Context, req *vtctldatapb.L
 				_, err = s.tmc.UpdateVReplicationWorkflow(ctx, tabletInfo.Tablet, &tabletmanagerdatapb.UpdateVReplicationWorkflowRequest{
 					Workflow: req.Name,
 					State:    ptr.Of(binlogdatapb.VReplicationWorkflowState_Stopped),
+					Message:  ptr.Of(Frozen),
 				})
 				if err != nil {
 					return vterrors.Wrapf(err, "failed to stop workflow %s on shard %s/%s", req.Name, tabletInfo.Keyspace, tabletInfo.Shard)
 				}
-				// Mark workflow as frozen.
-				query := fmt.Sprintf(SqlFreezeWorkflow, encodeString(tabletInfo.DbName()), encodeString(req.Name))
-				_, err = s.tmc.VReplicationExec(ctx, tabletInfo.Tablet, query)
-				return err
+				return nil
 			})
 			if err != nil {
 				return nil, err

go/vt/vttablet/tabletmanager/rpc_vreplication.go

@@ -61,7 +61,7 @@ const (
 	// Retrieve the current configuration values for a workflow's vreplication stream(s).
 	sqlSelectVReplicationWorkflowConfig = "select id, source, cell, tablet_types, state, message from %s.vreplication where workflow = %a"
 	// Update the configuration values for a workflow's vreplication stream.
-	sqlUpdateVReplicationWorkflowStreamConfig = "update %s.vreplication set state = %a, source = %a, cell = %a, tablet_types = %a %s where id = %a"
+	sqlUpdateVReplicationWorkflowStreamConfig = "update %s.vreplication set state = %a, source = %a, cell = %a, tablet_types = %a, message = %a %s where id = %a"
 	// Update field values for multiple workflows. The final format specifier is
 	// used to optionally add any additional predicates to the query.
 	sqlUpdateVReplicationWorkflows = "update /*vt+ ALLOW_UNSAFE_VREPLICATION_WRITE */ %s.vreplication set%s where db_name = '%s'%s"
@@ -568,6 +568,9 @@ func (tm *TabletManager) UpdateVReplicationWorkflow(ctx context.Context, req *ta
 		if !textutil.ValueIsSimulatedNull(req.TabletTypes) {
 			tabletTypes = req.TabletTypes
 		}
+		if req.Message != nil && *req.Message != sqltypes.Null.String() {
+			message = *req.Message
+		}
 		tabletTypesStr := topoproto.MakeStringTypeCSV(tabletTypes)
 		if req.TabletSelectionPreference != nil &&
 			((inorder && *req.TabletSelectionPreference == tabletmanagerdatapb.TabletSelectionPreference_UNKNOWN) ||
@@ -609,9 +612,10 @@ func (tm *TabletManager) UpdateVReplicationWorkflow(ctx context.Context, req *ta
 			"sc": sqltypes.StringBindVariable(string(source)),
 			"cl": sqltypes.StringBindVariable(strings.Join(cells, ",")),
 			"tt": sqltypes.StringBindVariable(tabletTypesStr),
+			"ms": sqltypes.StringBindVariable(message),
 			"id": sqltypes.Int64BindVariable(id),
 		}
-		parsed = sqlparser.BuildParsedQuery(sqlUpdateVReplicationWorkflowStreamConfig, sidecar.GetIdentifier(), ":st", ":sc", ":cl", ":tt", options, ":id")
+		parsed = sqlparser.BuildParsedQuery(sqlUpdateVReplicationWorkflowStreamConfig, sidecar.GetIdentifier(), ":st", ":sc", ":cl", ":tt", ":ms", options, ":id")
 		stmt, err = parsed.GenerateQuery(bindVars, nil)
 		if err != nil {
 			return nil, err

go/vt/vttablet/tabletmanager/rpc_vreplication_test.go

@@ -2062,11 +2062,8 @@ func TestExternalizeLookupVindex(t *testing.T) {
 						fmt.Sprintf("%d|%s||primary|Stopped|", vreplID, bls),
 					), nil)
 					targetTablet.vrdbClient.ExpectRequest(idQuery, idRes, nil)
-					targetTablet.vrdbClient.ExpectRequest(`update _vt.vreplication set state = 'Stopped', source = 'keyspace:"sourceks" shard:"0" filter:{rules:{match:"t1" filter:"select * from t1"}}', cell = '', tablet_types = '' where id in (1)`, &sqltypes.Result{}, nil)
+					targetTablet.vrdbClient.ExpectRequest(`update _vt.vreplication set state = 'Stopped', source = 'keyspace:"sourceks" shard:"0" filter:{rules:{match:"t1" filter:"select * from t1"}}', cell = '', tablet_types = '', message = 'FROZEN' where id in (1)`, &sqltypes.Result{}, nil)
 					targetTablet.vrdbClient.ExpectRequest(`select * from _vt.vreplication where id = 1`, streamsResult, nil)
-
-					freezeQuery := fmt.Sprintf(workflow.SqlFreezeWorkflow, sqltypes.EncodeStringSQL("vt_targetks"), sqltypes.EncodeStringSQL(tcase.request.Name))
-					tenv.tmc.setVReplicationExecResults(targetTablet.tablet, freezeQuery, &sqltypes.Result{})
 				}
 			}
 

proto/tabletmanagerdata.proto

@@ -776,6 +776,7 @@ message UpdateVReplicationWorkflowRequest {
   optional binlogdata.VReplicationWorkflowState state = 6;
   reserved 7; // unused, was: repeated string shards
   map<string, string> config_overrides = 8;
+  optional string message = 9;
 }
 
 message UpdateVReplicationWorkflowResponse {