VTGate VStream: Ensure reasonable delivery time for reshard journal e…

…vent (vitessio#16639) Signed-off-by: Matt Lord <[email protected]>
slackhq · Jan 16, 2025 · 0a23886 · 0a23886
1 parent bb6271b
commit 0a23886
Show file tree

Hide file tree

Showing 4 changed files with 497 additions and 16 deletions.
diff --git a/examples/local/vstream_client.go b/examples/local/vstream_client.go
@@ -23,13 +23,13 @@ import (
 	"log"
 	"time"
 
-	vtgatepb "vitess.io/vitess/go/vt/proto/vtgate"
-
-	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
-	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	_ "vitess.io/vitess/go/vt/vtctl/grpcvtctlclient"
 	_ "vitess.io/vitess/go/vt/vtgate/grpcvtgateconn"
 	"vitess.io/vitess/go/vt/vtgate/vtgateconn"
+
+	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
+	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
+	vtgatepb "vitess.io/vitess/go/vt/proto/vtgate"
 )
 
 /*
@@ -73,15 +73,18 @@ func main() {
 	}
 	defer conn.Close()
 	flags := &vtgatepb.VStreamFlags{
-		//MinimizeSkew:      false,
-		//HeartbeatInterval: 60, //seconds
+		// MinimizeSkew:      false,
+		// HeartbeatInterval: 60, //seconds
+		// StopOnReshard: true,
 	}
 	reader, err := conn.VStream(ctx, topodatapb.TabletType_PRIMARY, vgtid, filter, flags)
+	if err != nil {
+		log.Fatal(err)
+	}
 	for {
 		e, err := reader.Recv()
 		switch err {
 		case nil:
-			_ = e
 			fmt.Printf("%v\n", e)
 		case io.EOF:
 			fmt.Printf("stream ended\n")