VReplication: Add VTGate VStreamFlag to include journal events in the stream

examples/local/vstream_client.go

@@ -76,6 +76,7 @@ func main() {
 		// MinimizeSkew:      false,
 		// HeartbeatInterval: 60, //seconds
 		// StopOnReshard: true,
+		// IncludeReshardJournalEvents: true,
 	}
 	reader, err := conn.VStream(ctx, topodatapb.TabletType_PRIMARY, vgtid, filter, flags)
 	if err != nil {

go/test/endtoend/vreplication/vstream_test.go

@@ -598,7 +598,10 @@ func TestMultiVStreamsKeyspaceReshard(t *testing.T) {
 			Match: "/customer.*",
 		}},
 	}
-	flags := &vtgatepb.VStreamFlags{}
+	flags := &vtgatepb.VStreamFlags{
+		IncludeReshardJournalEvents: true,
+	}
+	journalEvents := 0
 
 	// Stream events but stop once we have a VGTID with positions for the old/original shards.
 	var newVGTID *binlogdatapb.VGtid
@@ -678,6 +681,9 @@ func TestMultiVStreamsKeyspaceReshard(t *testing.T) {
 						default:
 							require.FailNow(t, fmt.Sprintf("received event for unexpected shard: %s", shard))
 						}
+					case binlogdatapb.VEventType_JOURNAL:
+						require.True(t, ev.Journal.MigrationType == binlogdatapb.MigrationType_SHARDS)
+						journalEvents++
 					}
 				}
 			default:
@@ -694,6 +700,8 @@ func TestMultiVStreamsKeyspaceReshard(t *testing.T) {
 	// We should have a mix of events across the old and new shards.
 	require.Greater(t, oldShardRowEvents, 0)
 	require.Greater(t, newShardRowEvents, 0)
+	// We should have seen a reshard journal event.
+	require.Greater(t, journalEvents, 0)
 
 	// The number of row events streamed by the VStream API should match the number of rows inserted.
 	customerResult := execVtgateQuery(t, vtgateConn, ks, "select count(*) from customer")