Raise an IOError when source file in GcsIO.copy does not exist

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

Author: charlesccychen

google/cloud/dataflow/io/gcsio.py

@@ -55,6 +55,11 @@ def parse_gcs_path(gcs_path):
   return match.group(1), match.group(2)
 
 
+class GcsIOError(IOError, retry.PermanentException):
+  """GCS IO error that should not be retried."""
+  pass
+
+
 class GcsIO(object):
   """Google Cloud Storage I/O client."""
 
@@ -168,7 +173,15 @@ def copy(self, src, dest):
                                                 sourceObject=src_path,
                                                 destinationBucket=dest_bucket,
                                                 destinationObject=dest_path)
-    self.client.objects.Copy(request)
+    try:
+      self.client.objects.Copy(request)
+    except HttpError as http_error:
+      if http_error.status_code == 404:
+        # This is a permanent error that should not be retried.  Note that
+        # FileSink.finalize_write expects an IOError when the source file does
+        # not exist.
+        raise GcsIOError(errno.ENOENT, 'Source file not found: %s' % src)
+      raise
 
   # We intentionally do not decorate this method with a retry, since the
   # underlying copy and delete operations are already idempotent operations

google/cloud/dataflow/io/gcsio_test.py

@@ -109,7 +109,9 @@ def Insert(self, insert_request, upload=None):  # pylint: disable=invalid-name
   def Copy(self, copy_request):  # pylint: disable=invalid-name
     src_file = self.get_file(copy_request.sourceBucket,
                              copy_request.sourceObject)
-    assert src_file is not None
+    if not src_file:
+      raise HttpError(httplib2.Response({'status': '404'}), '404 Not Found',
+                      'https://fake/url')
     generation = self.get_last_generation(copy_request.destinationBucket,
                                           copy_request.destinationObject) + 1
     dest_file = FakeFile(copy_request.destinationBucket,
@@ -219,6 +221,10 @@ def test_copy(self):
     self.assertTrue(gcsio.parse_gcs_path(dest_file_name) in
                     self.client.objects.files)
 
+    self.assertRaises(IOError, self.gcs.copy,
+                      'gs://gcsio-test/non-existent',
+                      'gs://gcsio-test/non-existent-destination')
+
   def test_copytree(self):
     src_dir_name = 'gs://gcsio-test/source/'
     dest_dir_name = 'gs://gcsio-test/dest/'