apache
diff --git a/‎sdks/python/apache_beam/io/iobase.py
Lines changed: 13 additions & 8 deletions b/‎sdks/python/apache_beam/io/iobase.py
Lines changed: 13 additions & 8 deletions
diff --git a/‎sdks/python/apache_beam/transforms/combiners.py
Lines changed: 9 additions & 9 deletions b/‎sdks/python/apache_beam/transforms/combiners.py
Lines changed: 9 additions & 9 deletions
@@ -998,7 +998,7 @@ def expand(self, pcoll):
     # type: (pvalue.PCollection[InT]) -> pvalue.PCollection[None]
     do_once = pcoll.pipeline | 'DoOnce' >> core.Create([None])
     init_result_coll = do_once | 'InitializeWrite' >> core.Map(
-        lambda _, sink: sink.initialize_write(), self.sink)
+        lambda _, sink=self.sink: sink.initialize_write())
     if getattr(self.sink, 'num_shards', 0):
       min_shards = self.sink.num_shards
       if min_shards == 1:
@@ -1029,9 +1029,9 @@ def getitem(x):
     # fused.
     pre_finalize_coll = do_once | 'PreFinalize' >> core.FlatMap(
         _pre_finalize,
-        self.sink,
-        AsSingleton(init_result_coll),
-        AsIter(write_result_coll))
+        sink=self.sink,
+        init_result=AsSingleton(init_result_coll),
+        write_results=AsIter(write_result_coll))
     return do_once | 'FinalizeWrite' >> core.FlatMap(
         _finalize_write,
         self.sink,
@@ -1084,12 +1084,17 @@ def process(self, element, init_result):
     return [window.TimestampedValue(writer.close(), timestamp.MAX_TIMESTAMP)]
 
 
-def _pre_finalize(unused_element, sink, init_result, write_results):
+# FIXME: when we migrate to python3-only, change this to:
+#  def _pre_finalize(unused_element, *, sink, init_result, write_results):
+def _pre_finalize(unused_element, sink=None, init_result=None,
+                  write_results=None):
   return sink.pre_finalize(init_result, write_results)
 
-
-def _finalize_write(unused_element, sink, init_result, write_results,
-                    min_shards, pre_finalize_results):
+# FIXME: when we migrate to python3-only, change this to:
+#  def _finalize_write(unused_element, *, sink, init_result, ...):
+def _finalize_write(unused_element, sink=None, init_result=None,
+                    write_results=None, min_shards=None,
+                    pre_finalize_results=None):
   write_results = list(write_results)
   extra_shards = []
   if len(write_results) < min_shards:
 
@@ -401,9 +401,9 @@ def __init__(self, n, less_than, key):
     self._less_than = None if less_than is operator.lt else less_than
     self._key = key
 
-  def process(self, key_and_bundles):
-    # type: (Tuple[None, Iterable[List[T]]]) -> Iterator[List[T]]
-    _, bundles = key_and_bundles
+  def process(self, element, *args, **kwargs):
+    # type: (Tuple[None, Iterable[List[T]]], *Any, **Any) -> Iterator[List[T]]
+    _, bundles = element
 
     def push(hp, e):
       if len(hp) < self._n:
@@ -422,12 +422,12 @@ def push(hp, e):
       for bundle in bundles:
         if not heapc:
           heapc = [
-            cy_combiners.ComparableValue(element, self._less_than, self._key)
-            for element in bundle]
+            cy_combiners.ComparableValue(item, self._less_than, self._key)
+            for item in bundle]
           continue
-        for element in reversed(bundle):
+        for item in reversed(bundle):
           if push(heapc, cy_combiners.ComparableValue(
-              element, self._less_than, self._key)):
+              item, self._less_than, self._key)):
             break
       heapc.sort()
       yield [wrapper.value for wrapper in reversed(heapc)]
@@ -438,8 +438,8 @@ def push(hp, e):
         if not heap:
           heap = bundle
           continue
-        for element in reversed(bundle):
-          if push(heap, element):
+        for item in reversed(bundle):
+          if push(heap, item):
             break
       heap.sort()
       yield heap[::-1]