Backport "Correctly test os.stat().st_size"

apache/airflow#3843

Change-Id: I95e09cb68a2b813c2eeb487ee4e1707c9eb92dd5
GitOrigin-RevId: dbd8882bf86297d285a8fc5549ce80da0d8a81a4

Author: Cloud Composer Team

airflow/contrib/operators/dataflow_operator.py

@@ -17,6 +17,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
+import os
 import re
 import uuid
 import copy
@@ -358,7 +359,7 @@ def google_cloud_to_local(self, file_name):
         # Extracts bucket_id and object_id by first removing 'gs://' prefix and
         # then split the remaining by path delimiter '/'.
         path_components = file_name[self.GCS_PREFIX_LENGTH:].split('/')
-        if path_components < 2:
+        if len(path_components) < 2:
             raise Exception(
                 'Invalid Google Cloud Storage (GCS) object path: {}.'
                 .format(file_name))
@@ -369,7 +370,7 @@ def google_cloud_to_local(self, file_name):
                                                  path_components[-1])
         file_size = self._gcs_hook.download(bucket_id, object_id, local_file)
 
-        if file_size > 0:
+        if os.stat(local_file).st_size > 0:
             return local_file
         raise Exception(
             'Failed to download Google Cloud Storage GCS object: {}'