Do not panic if an element has no windows during PersistBundle. (#36610)

* Do not panic if an element has no windows during PersistBundle.

* Add a test for this case.

Author: shunping

sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go

@@ -791,8 +791,7 @@ func reElementResiduals(residuals []Residual, inputInfo PColInfo, rb RunBundle)
 			panic("error decoding residual header:" + err.Error())
 		}
 		if len(ws) == 0 {
-			slog.Error("reElementResiduals: sdk provided a windowed value header 0 windows", "bundle", rb)
-			panic("error decoding residual header: sdk provided a windowed value header 0 windows")
+			slog.Warn("reElementResiduals: sdk provided a windowed value header 0 windows", "bundle", rb)
 		}
 		// POSSIBLY BAD PATTERN: The buffer is invalidated on the next call, which doesn't always happen.
 		// But the decoder won't be mutating the buffer bytes, just reading the data. So the elmBytes
@@ -852,8 +851,7 @@ func (em *ElementManager) PersistBundle(rb RunBundle, col2Coders map[string]PCol
 					panic("error decoding watermarks")
 				}
 				if len(ws) == 0 {
-					slog.Error("PersistBundle: sdk provided a windowed value header 0 windows", "bundle", rb)
-					panic("error decoding residual header: sdk provided a windowed value header 0 windows")
+					slog.Warn("PersistBundle: sdk provided a windowed value header 0 windows", "bundle", rb)
 				}
 				// TODO: Optimize unnecessary copies. This is doubleteeing.
 				elmBytes := info.EDec(tee)

sdks/python/apache_beam/transforms/window_test.py

@@ -192,6 +192,19 @@ def test_sliding_windows(self):
                   ('key @ [2.0, 6.0)', [2, 3])]
       assert_that(result, equal_to(expected))
 
+  def test_sliding_windows_period_longer_than_size(self):
+    with TestPipeline() as p:
+      pcoll = self.timestamped_key_values(p, 'key', 1, 2, 3, 4, 5, 6, 7, 8)
+      result = (
+          pcoll
+          | 'w' >> WindowInto(SlidingWindows(period=4, size=2))
+          | GroupByKey()
+          | beam.MapTuple(lambda k, vs: (k, sorted(vs)))
+          | beam.ParDo(ReifyWindowsFn()))
+      expected = [('key @ [0.0, 2.0)', [1]), ('key @ [4.0, 6.0)', [4, 5]),
+                  ('key @ [8.0, 10.0)', [8])]
+      assert_that(result, equal_to(expected))
+
   def test_sessions(self):
     with TestPipeline() as p:
       pcoll = self.timestamped_key_values(p, 'key', 1, 2, 3, 20, 35, 27)