{"record":{"id":"341d260cae164f2f","repo":"risingwavelabs/risingwave","slug":"input-stream-terminated-unexpectedly-during-normal","errorCode":null,"errorMessage":"Input stream terminated unexpectedly during normal ingestion","messagePattern":"Input stream terminated unexpectedly during normal ingestion","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/mview/materialize.rs","lineNumber":619,"sourceCode":"\n                                        yield Message::Chunk(chunk);\n                                    }\n                                }\n                            }\n                            Message::Barrier(barrier) => {\n                                *inner_state = MaterializeStreamState::CommitAndYieldBarrier {\n                                    barrier,\n                                    expect_next_state: Box::new(\n                                        MaterializeStreamState::NormalIngestion,\n                                    ),\n                                };\n                                continue 'main_loop;\n                            }\n                        }\n                    }\n\n                    return Err(StreamExecutorError::from(ErrorKind::Uncategorized(\n                        anyhow::anyhow!(\n                            \"Input stream terminated unexpectedly during normal ingestion\"\n                        ),\n                    )));\n                }\n                MaterializeStreamState::MergingData => {\n                    let Some(refresh_args) = self.refresh_args.as_mut() else {\n                        panic!(\n                            \"MaterializeExecutor entered CleanUp state without refresh_args configured\"\n                        );\n                    };\n                    tracing::info!(table_id = %refresh_args.table_id, \"on_load_finish: Starting table replacement operation\");\n\n                    debug_assert_eq!(\n                        self.state_table.vnodes(),\n                        refresh_args.staging_table.vnodes()\n                    );\n                    debug_assert_eq!(\n                        refresh_args.staging_table.vnodes(),","sourceCodeStart":601,"sourceCodeEnd":637,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/mview/materialize.rs#L601-L637","documentation":"The materialize executor keeps buffering upstream data in `MaterializeStreamState::Buffering` while waiting for a barrier that lets it switch to normal ingestion. If, while in normal ingestion, the upstream channel yields `None` (stream end) instead of a barrier/chunk, the executor treats it as a fatal topology violation and throws this uncategorized error, because the input to a materialized-view executor must never end during normal operation.","triggerScenarios":"The upstream actor/fragment terminates (crashes, is cancelled by a DDL like DROP, or the meta node fails over the fragment) while the materialize executor is in normal ingestion, so the channel closes rather than delivering a `Stop`/barrier message.","commonSituations":"Concurrent DROP MATERIALIZED VIEW / schema change while the MV is actively ingesting; upstream actor panic or OOM kill during mv creation backfill completion; meta failover bugs that tear down fragments without sending proper barriers.","solutions":["Check the upstream actor logs for a crash, panic, or explicit cancellation and fix the root cause (often OOM or a meta-service failover bug)","Retry the MV creation or refresh the materialized view if the termination came from a transient cluster event","Check whether a concurrent DDL statement (DROP/ALTER) raced with the job; serialize DDL against running MV creation","If reproducible without external cause, file an issue — a healthy pipeline must terminate executors via barriers, not channel closure"],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// Before relying on a long-lived MV, ensure no DDL targets it concurrently\nassert_no_concurrent_ddl(materialized_view_name)?;","typeGuard":null,"tryCatchPattern":"match ingest_result {\n    Err(e) if e.to_string().contains(\"Input stream terminated unexpectedly\") => {\n        // Upstream actor died: check upstream logs, then retry after root cause is cleared\n        inspect_upstream_actor_logs();\n        retry_with_backoff(create_materialized_view);\n    }\n    other => other?,\n}","preventionTips":["Avoid running DROP/ALTER DDL against a table or MV while downstream MV creation is in progress","Monitor upstream actor memory (OOM kills terminate actors without barriers)","Ensure meta failover procedures are tested; unclean failover tears down channels","Pin cluster to a RisingWave version without known fragment-teardown bugs"],"tags":["rust","streaming","materialized-view","upstream-failure"],"backgroundTag":"upstream-api-error","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}