{"record":{"id":"f5f0def82a96faeb","repo":"vitessio/vitess","slug":"vstreamer-is-not-open-f5f0de","errorCode":null,"errorMessage":"VStreamer is not open","messagePattern":"VStreamer is not open","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"go/vt/vttablet/tabletserver/vstreamer/tablestreamer.go","lineNumber":157,"sourceCode":"\tlog.Info(fmt.Sprintf(\"Found %d tables to stream: %s\", len(ts.tables), strings.Join(ts.tables, \", \")))\n\tfor _, tableName := range ts.tables {\n\t\tlog.Info(\"Streaming table \" + tableName)\n\t\tif err := ts.streamTable(ts.ctx, tableName); err != nil {\n\t\t\tlog.Error(fmt.Sprintf(\"Streaming table %s failed: %v\", tableName, err))\n\t\t\treturn err\n\t\t}\n\t\tlog.Info(\"Finished streaming table \" + tableName)\n\t}\n\tlog.Info(fmt.Sprintf(\"Finished streaming %d tables\", len(ts.tables)))\n\treturn nil\n}\n\nfunc (ts *tableStreamer) newRowStreamer(ctx context.Context, query string, lastpk []sqltypes.Value,\n\tsend func(*binlogdatapb.VStreamRowsResponse) error,\n) (*rowStreamer, func(), error) {\n\tvse := ts.vse\n\tif atomic.LoadInt32(&vse.isOpen) == 0 {\n\t\treturn nil, nil, errors.New(\"VStreamer is not open\")\n\t}\n\tvse.mu.Lock()\n\tdefer vse.mu.Unlock()\n\n\trowStreamer := newRowStreamer(ctx, vse.env.Config().DB.FilteredWithDB(), vse.se, query, lastpk, vse.lvschema,\n\t\tsend, vse, RowStreamerModeAllTables, ts.snapshotConn, ts.options)\n\n\tidx := vse.streamIdx\n\tvse.rowStreamers[idx] = rowStreamer\n\tvse.streamIdx++\n\t// Now that we've added the stream, increment wg.\n\t// This must be done before releasing the lock.\n\tvse.wg.Add(1)\n\n\t// Remove stream from map and decrement wg when it ends.\n\tcancel := func() {\n\t\tvse.mu.Lock()\n\t\tdefer vse.mu.Unlock()","sourceCodeStart":139,"sourceCodeEnd":175,"githubUrl":"https://github.com/vitessio/vitess/blob/01a25a7d176f94613b8d59d799f438380a8760e4/go/vt/vttablet/tabletserver/vstreamer/tablestreamer.go#L139-L175","documentation":"tableStreamer.newRowStreamer re-checks the VStreamer engine's isOpen flag before spawning each per-table row streamer. Because table streaming is long-lived, the engine may have been closed (or not yet fully open) between stream creation and this per-row-stream creation; when isOpen is 0 it returns \"VStreamer is not open\".","triggerScenarios":"A running tableStreamer (streamTable / multi-table copy) calling newRowStreamer after Engine.Close, or before the engine finished opening; recursively invoked as newRowStreamer creates child row streamers.","commonSituations":"vttablet shutdown or tabletserver restart while a VReplication copy phase is actively streaming tables; cancelling VReplication workflows racing stream creation.","solutions":["Treat as a benign shutdown signal: the stream should be cancelled/restarted — VReplication workflows will retry automatically.","If seen outside shutdowns, verify the tabletserver/VStreamer engine stayed open (check startup/shutdown logs).","Retry the VStream/workflow once the tablet is fully healthy."],"exampleFix":"// before\nrs, cancel, err := ts.newRowStreamer(ctx, query, lastpk, send) // fails mid-shutdown\n// after\nif err != nil && err.Error() == \"VStreamer is not open\" { return gracefulStop() } // handle engine shutdown","handlingStrategy":"try-catch","validationCode":null,"typeGuard":"func engineShutdownErr(err error) bool { return err != nil && strings.Contains(err.Error(), \"VStreamer is not open\") }","tryCatchPattern":"rs, cancel, err := ts.newRowStreamer(ctx, query, lastpk, send)\nif engineShutdownErr(err) {\n  log.Info(\"VStreamer engine closed mid-stream; stopping table stream gracefully\")\n  return nil, nil, nil\n}","preventionTips":["Expect this during vttablet shutdown; ensure VReplication retry logic handles it.","Drain/cancel VReplication workflows before planned restarts.","Monitor tablet restarts correlated with VReplication error rates."],"tags":["vitess","vstreamer","vreplication","lifecycle"],"backgroundTag":"service-not-initialized","analyzedSha":"01a25a7d176f94613b8d59d799f438380a8760e4","analyzedAt":"2026-09-01T17:28:30.605Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}