PILOT-2688: Skip the file with size 0

.gitignore

@@ -155,3 +155,4 @@ integration_tests
 # cli manifest data
 ./manifest.json
 manifest.json
+test

app/services/file_manager/file_upload/file_upload.py

@@ -165,7 +165,14 @@ def simple_upload(  # noqa: C901
         # first remove the input path from the file path
         file_path_sub = file.replace(input_path + '/', '')
         object_path = os.path.join(target_folder, file_path_sub)
-        file_objects.append(FileObject(object_path, file))
+
+        # generate a placeholder for each file
+        file_object = FileObject(object_path, file)
+        # skip the file with 0 size
+        if file_object.total_size == 0:
+            logger.warning(f'Skip the file with 0 size: {file_object.file_name}')
+        else:
+            file_objects.append(FileObject(object_path, file))
 
     # here add the batch of 500 per loop, the pre upload api cannot
     # process very large amount of file at same time. otherwise it will timeout
@@ -190,10 +197,11 @@ def simple_upload(  # noqa: C901
     pool = ThreadPool(num_of_thread + 1)
     pool.apply_async(upload_client.upload_token_refresh)
     on_success_res = []
+
+    file_object: FileObject
     for file_object in pre_upload_infos:
         chunk_res = upload_client.stream_upload(file_object, pool)
-        # NOTE: if there is some racing error make the combine chunks
-        # out of thread pool.
+        # the on_success api will be called after all chunk uploaded
         res = pool.apply_async(
             upload_client.on_succeed,
             args=(file_object, tags, chunk_res),
@@ -219,7 +227,7 @@ def simple_upload(  # noqa: C901
             upload_client.create_file_lineage(source_file)
             os.remove(file_batchs[0]) if os.path.isdir(input_path) and job_type == UploadType.AS_FILE else None
 
-    num_of_file = len(upload_file_path)
+    num_of_file = len(pre_upload_infos)
     logger.info(f'Upload Time: {time.time() - upload_start_time:.2f}s for {num_of_file:d} files')
 
     return [file_object.item_id for file_object in pre_upload_infos]

app/services/file_manager/file_upload/upload_client.py

@@ -364,9 +364,7 @@ def on_succeed(self, file_object: FileObject, tags: List[str], chunk_result: Lis
         """
 
         # check if all the chunks have been uploaded
-        for res in chunk_result:
-            while res.get() is None:
-                time.sleep(1)
+        [res.wait() for res in chunk_result]
 
         for i in range(AppConfig.Env.resilient_retry):
             url = self.base_url + '/v1/files'

tests/app/services/file_manager/file_upload/test_file_upload.py

@@ -109,6 +109,21 @@ def test_assemble_path_at_non_existing_folder(mocker):
     assert create_folder_flag is True
 
 
+def test_file_upload_skip_empty_file(mocker):
+    file_name = 'test'
+    upload_event = {
+        'file': file_name,
+        'project_code': 'test_project',
+        'zone': 'greenroom',
+    }
+
+    mocker.patch('os.path.isdir', return_value=False)
+    mocker.patch('app.services.file_manager.file_upload.models.FileObject.generate_meta', return_value=(0, 0))
+
+    item_ids = simple_upload(upload_event)
+    assert len(item_ids) == 0
+
+
 def test_dont_allow_tagging_when_folder_upload(mocker, capfd):
     file_name = 'test'
     upload_event = {