Fixes a bug in progress reporting in TextFileReader

It was 1) thread-unsafe (accessed a reader's variable from a different
thread) and 2) incorrect (did not return position of the last returned
record).

Because of this, the reader could end up in a situation where it would
keep returning a position past the end of the range, confusing dynamic
work rebalancing.

----Release Notes----
[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=119983487

Author: jkff

google/cloud/dataflow/internal/windmill_pb2.py

@@ -1292,7 +1292,7 @@
     _descriptor.FieldDescriptor(
       name='double_scalar', full_name='windmill.Counter.double_scalar', index=2,
       number=3, type=1, cpp_type=5, label=1,
-      has_default_value=False, default_value=0,
+      has_default_value=False, default_value=float(0),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       options=None),

google/cloud/dataflow/io/fileio.py

@@ -313,8 +313,8 @@ def __iter__(self):
       yield self.source.coder.decode(line)
 
   def get_progress(self):
-    return iobase.ReaderProgress(
-        position=iobase.ReaderPosition(byte_offset=self.current_offset))
+    return iobase.ReaderProgress(position=iobase.ReaderPosition(
+        byte_offset=self.range_tracker.last_record_start))
 
   def request_dynamic_split(self, dynamic_split_request):
     assert dynamic_split_request is not None

google/cloud/dataflow/io/fileio_test.py

@@ -49,10 +49,7 @@ def progress_with_offsets(self, input_lines,
         start_offset=start_offset, end_offset=end_offset)
     progress_record = []
     with source.reader() as reader:
-      # Starting value of percent_complete might be larger than zero since we
-      # will ignore the first line of a record if the starting position of the
-      # line is smaller than the start offset of the source.
-      self.assertGreaterEqual(reader.get_progress().position.byte_offset, 0)
+      self.assertEqual(reader.get_progress().position.byte_offset, -1)
       for line in reader:
         self.assertIsNotNone(line)
         progress_record.append(reader.get_progress().position.byte_offset)
@@ -78,14 +75,14 @@ def test_progress_entire_file(self):
         file_path=self.create_temp_file('\n'.join(lines)))
     progress_record = []
     with source.reader() as reader:
-      self.assertEqual(0, reader.get_progress().position.byte_offset)
+      self.assertEqual(-1, reader.get_progress().position.byte_offset)
       for line in reader:
         self.assertIsNotNone(line)
         progress_record.append(reader.get_progress().position.byte_offset)
-      self.assertEqual(18, reader.get_progress().position.byte_offset)
+      self.assertEqual(13, reader.get_progress().position.byte_offset)
 
     self.assertEqual(len(progress_record), 3)
-    self.assertEqual(progress_record, [6, 13, 18])
+    self.assertEqual(progress_record, [0, 6, 13])
 
   def try_splitting_reader_at(self, reader, split_request, expected_response):
     actual_response = reader.request_dynamic_split(split_request)