Fix logs downloading for tasks (apache#63)

(cherry picked from commit e979162)
(cherry picked from commit 87f2abb)
(cherry picked from commit 06f19c9)
(cherry picked from commit 1b52c3a)

airflow/utils/log/es_task_handler.py

@@ -144,10 +144,10 @@ def _read(self, ti, try_number, metadata=None):
         if 'last_log_timestamp' in metadata:
             last_log_ts = timezone.parse(metadata['last_log_timestamp'])
             if cur_ts.diff(last_log_ts).in_minutes() >= 5 or 'max_offset' in metadata \
-                    and offset >= metadata['max_offset']:
+                    and int(offset) >= int(metadata['max_offset']):
                 metadata['end_of_log'] = True
 
-        if offset != next_offset or 'last_log_timestamp' not in metadata:
+        if int(offset) != int(next_offset) or 'last_log_timestamp' not in metadata:
             metadata['last_log_timestamp'] = str(cur_ts)
 
         # If we hit the end of the log, remove the actual end_of_log message