[BEAM-9945] Ensure that the read index represents the number of fully processed elements including at the end of the channel or after splitting.

Author: Kyle Weaver

sdks/python/apache_beam/runners/worker/bundle_processor.py

@@ -191,14 +191,14 @@ def __init__(self,
             self.windowed_coder)
     ]
     self.splitting_lock = threading.Lock()
+    self.index = -1
+    self.stop = float('inf')
     self.started = False
 
   def start(self):
     # type: () -> None
     super(DataInputOperation, self).start()
     with self.splitting_lock:
-      self.index = -1
-      self.stop = float('inf')
       self.started = True
 
   def process(self, windowed_value):
@@ -317,8 +317,15 @@ def is_valid_split_point(index):
   def finish(self):
     # type: () -> None
     with self.splitting_lock:
+      self.index += 1
       self.started = False
 
+  def reset(self):
+    # type: () -> None
+    self.index = -1
+    self.stop = float('inf')
+    super(DataInputOperation, self).reset()
+
 
 class _StateBackedIterable(object):
   def __init__(self,