VReplication: Add VTGate VStreamFlag to include journal events in the stream (vitessio#16737)

Signed-off-by: Malcolm Akinje <malcolm.akinje@gmail.com>
Signed-off-by: Malcolm Akinje <makinje@slack-corp.com>

Author: makinje16

examples/local/vstream_client.go

@@ -75,6 +75,7 @@ func main() {
 	flags := &vtgatepb.VStreamFlags{
 		//MinimizeSkew:      false,
 		//HeartbeatInterval: 60, //seconds
+		// IncludeReshardJournalEvents: true,
 	}
 	reader, err := conn.VStream(ctx, topodatapb.TabletType_PRIMARY, vgtid, filter, flags)
 	for {

go/test/endtoend/vreplication/vstream_test.go

@@ -601,7 +601,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
@@ -683,6 +686,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:
@@ -697,8 +703,10 @@ func TestMultiVStreamsKeyspaceReshard(t *testing.T) {
 	}()
 
 	// We should have a mix of events across the old and new shards.
-	require.NotZero(t, oldShardRowEvents)
-	require.NotZero(t, newShardRowEvents)
+	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")