From c0e1695b8757a3d67cd8e0722c38be724775bd5e Mon Sep 17 00:00:00 2001
From: MOZGIII <mike-n@narod.ru>
Date: Sun, 28 Jun 2020 10:45:50 +0300
Subject: [PATCH] Handle data parsing errors as stream ends

Signed-off-by: MOZGIII <mike-n@narod.ru>
---
 src/kubernetes/stream.rs | 24 ++++++++++++++++++++++++
 1 file changed, 24 insertions(+)

diff --git a/src/kubernetes/stream.rs b/src/kubernetes/stream.rs
index 8b280073705264..b1f607153ef4d5 100644
--- a/src/kubernetes/stream.rs
+++ b/src/kubernetes/stream.rs
@@ -30,6 +30,16 @@ where
             let responses = decoder.process_next_chunk(chunk.as_ref());
             emit!(internal_events::ChunkProcessed{ byte_size: chunk.len() });
             for response in responses {
+                // Sometimes Kubernetes API starts returning `null`s in
+                // the object field while streaming the response.
+                // Handle it as if the stream has ended.
+                // See https://github.com/kubernetes/client-go/issues/334
+                if let Err(ResponseError::Json(error)) = &response {
+                    if error.is_data() {
+                        warn!(message = "handling response json parsing data error as steram end", ?error);
+                        return;
+                    }
+                }
                 let response = response.context(Parsing)?;
                 yield response;
             }
@@ -166,4 +176,18 @@ mod tests {
             assert!(out_stream.next().await.is_none());
         })
     }
+
+    #[test]
+    fn test_sudden_null() {
+        test_util::trace_init();
+        test_util::block_on_std(async move {
+            let chunks: Vec<Result<_, std::io::Error>> = vec![Ok("null")];
+            let sample_body = hyper_body_from_chunks(chunks);
+
+            let out_stream = body::<_, WatchResponse<Pod>>(sample_body);
+            pin_mut!(out_stream);
+
+            assert!(out_stream.next().await.is_none());
+        })
+    }
 }