Updates MapTask._parse_avro_source() so that start and end position of sources are set properly.

Also does a small logging update.

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

Author: chamikaramj

google/cloud/dataflow/io/range_trackers.py

@@ -219,8 +219,8 @@ def try_split_at_position(self, decoded_split_position):
                       'of range', self, decoded_split_position)
         return False
 
-      logging.info('Agreeing to split %r at %r'
-                   , self, decoded_split_position)
+      logging.debug('Agreeing to split %r at %r'
+                    , self, decoded_split_position)
       self._decoded_stop_pos = decoded_split_position
       return True
 

google/cloud/dataflow/worker/maptask.py

@@ -678,10 +678,16 @@ def _parse_avro_source(specs, unused_codec_specs, unused_context):
       # Note that the worker does not really implement AVRO yet.It takes
       # advantage that both reading and writing is done through the worker to
       # choose a supported format (text files with one pickled object per line).
+      start_offset = None
+      if 'start_offset' in specs:
+        start_offset = int(specs['start_offset']['value'])
+      end_offset = None
+      if 'end_offset' in specs:
+        end_offset = int(specs['end_offset']['value'])
       return io.TextFileSource(
           file_path=specs['filename']['value'],
-          start_offset=None,
-          end_offset=None,
+          start_offset=start_offset,
+          end_offset=end_offset,
           strip_trailing_newlines=True,
           coder=coders.Base64PickleCoder())