{"record":{"id":"b3447693011461c2","repo":"vectordotdev/vector","slug":"failed-type-coercion-self-is-not-a-stream","errorCode":null,"errorMessage":"Failed type coercion, {self:?} is not a Stream","messagePattern":"Failed type coercion, (.+?) is not a Stream","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"lib/vector-core/src/sink.rs","lineNumber":61,"sourceCode":"    /// # Panics\n    ///\n    /// This function will panic if the self instance is not `VectorSink::Sink`.\n    pub fn into_sink(self) -> Box<dyn Sink<EventArray, Error = ()> + Send + Unpin> {\n        match self {\n            Self::Sink(sink) => sink,\n            _ => panic!(\"Failed type coercion, {self:?} is not a Sink\"),\n        }\n    }\n\n    /// Converts `VectorSink` into a `StreamSink`\n    ///\n    /// # Panics\n    ///\n    /// This function will panic if the self instance is not `VectorSink::Stream`.\n    pub fn into_stream(self) -> Box<dyn StreamSink<EventArray> + Send> {\n        match self {\n            Self::Stream(stream) => stream,\n            _ => panic!(\"Failed type coercion, {self:?} is not a Stream\"),\n        }\n    }\n\n    /// Converts an event sink into a `VectorSink`\n    ///\n    /// Deprecated in favor of `VectorSink::from_event_streamsink`. See [vector/9261]\n    /// for more info.\n    ///\n    /// [vector/9261]: https://github.com/vectordotdev/vector/issues/9261\n    #[deprecated]\n    pub fn from_event_sink(sink: impl Sink<Event, Error = ()> + Send + Unpin + 'static) -> Self {\n        VectorSink::Sink(Box::new(EventSink::new(sink)))\n    }\n\n    /// Converts an event stream into a `VectorSink`\n    pub fn from_event_streamsink(sink: impl StreamSink<Event> + Send + 'static) -> Self {\n        let sink = Box::new(sink);\n        VectorSink::Stream(Box::new(EventStream { sink }))","sourceCodeStart":43,"sourceCodeEnd":79,"githubUrl":"https://github.com/vectordotdev/vector/blob/3708c39b12a93212ed8b8d7510b4cc7769cb5864/lib/vector-core/src/sink.rs#L43-L79","documentation":"into_stream() is the inverse coercion on VectorSink: it unwraps the Stream variant (a boxed StreamSink<EventArray>) and panics if the value is the Sink variant. Like into_sink(), the panic is documented and indicates the caller assumed the wrong sink shape at runtime.","triggerScenarios":"Calling sink.into_stream() on a VectorSink built as VectorSink::Sink (e.g. via from_event_sink or a sink adapter wrapping a futures::Sink), usually in stream-driving wrapper code.","commonSituations":"Wrapper code written for the StreamSink model receiving a legacy or custom sink that wraps futures::Sink; test harnesses that consume every sink as a stream.","solutions":["Branch on the variant: use into_stream() only for VectorSink::Stream, into_sink() for VectorSink::Sink","Prefer VectorSink::run(input) so the enum is consumed without assuming either shape","If a StreamSink is required downstream, wrap the futures::Sink with an adapter that implements StreamSink instead of coercing"],"exampleFix":"// before\nlet stream_sink = vector_sink.into_stream(); // panics for Sink variant\n\n// after\nif matches!(vector_sink, VectorSink::Stream(_)) {\n    let stream_sink = vector_sink.into_stream();\n} else {\n    let fut_sink = vector_sink.into_sink();\n}","handlingStrategy":"type-guard","validationCode":"if matches!(vector_sink, VectorSink::Stream(_)) {\n    let s = vector_sink.into_stream(); // safe\n} else {\n    let s = vector_sink.into_sink();\n}","typeGuard":"fn is_stream_variant(v: &VectorSink) -> bool {\n    matches!(v, VectorSink::Stream(_))\n}","tryCatchPattern":null,"preventionTips":["Consume sinks through VectorSink::run unless you specifically need one shape","When a StreamSink is required, adapt futures::Sink implementations instead of coercing","Check the variant in debug_assertions to catch shape mismatches in tests"],"tags":["rust","vector","sink","type-coercion","panic"],"backgroundTag":"invalid-type-coercion","analyzedSha":"3708c39b12a93212ed8b8d7510b4cc7769cb5864","analyzedAt":"2026-08-20T07:02:18.786Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T15:18:49.778Z"}