Problem with Wasb v12 remote logging when blob already exists

[flolas]

Apache Airflow version: 2.02

Kubernetes version (if you are using kubernetes) (use kubectl version): v1.20.5

Environment:

• Cloud provider or hardware configuration: AKS

What happened:

When using wasb for remote logging and backfilling a DAG, if the blob name already exists in the bucket, pods fails

> `Error in atexit._run_exitfuncs:
> Traceback (most recent call last):
>   File "/home/airflow/.local/lib/python3.6/site-packages/azure/storage/blob/_upload_helpers.py", line 105, in upload_block_blob
>     **kwargs)
>   File "/home/airflow/.local/lib/python3.6/site-packages/azure/storage/blob/_generated/operations/_block_blob_operations.py", line 231, in upload
>     map_error(status_code=response.status_code, response=response, error_map=error_map)
>   File "/home/airflow/.local/lib/python3.6/site-packages/azure/core/exceptions.py", line 102, in map_error
>     raise error
> azure.core.exceptions.ResourceExistsError: Operation returned an invalid status 'The specified blob already exists.'
> 
> During handling of the above exception, another exception occurred:
> 
> Traceback (most recent call last):
>   File "/usr/local/lib/python3.6/logging/__init__.py", line 1946, in shutdown
>     h.close()
>   File "/home/airflow/.local/lib/python3.6/site-packages/airflow/providers/microsoft/azure/log/wasb_task_handler.py", line 103, in close
>     self.wasb_write(log, remote_loc, append=True)
>   File "/home/airflow/.local/lib/python3.6/site-packages/airflow/providers/microsoft/azure/log/wasb_task_handler.py", line 192, in wasb_write
>     remote_log_location,
>   File "/home/airflow/.local/lib/python3.6/site-packages/airflow/providers/microsoft/azure/hooks/wasb.py", line 217, in load_string
>     self.upload(container_name, blob_name, string_data, **kwargs)
>   File "/home/airflow/.local/lib/python3.6/site-packages/airflow/providers/microsoft/azure/hooks/wasb.py", line 274, in upload
>     return blob_client.upload_blob(data, blob_type, length=length, **kwargs)
>   File "/home/airflow/.local/lib/python3.6/site-packages/azure/core/tracing/decorator.py", line 83, in wrapper_use_tracer
>     return func(*args, **kwargs)
>   File "/home/airflow/.local/lib/python3.6/site-packages/azure/storage/blob/_blob_client.py", line 685, in upload_blob
>     return upload_block_blob(**options)
>   File "/home/airflow/.local/lib/python3.6/site-packages/azure/storage/blob/_upload_helpers.py", line 157, in upload_block_blob
>     process_storage_error(error)
>   File "/home/airflow/.local/lib/python3.6/site-packages/azure/storage/blob/_shared/response_handlers.py", line 150, in process_storage_error
>     error.raise_with_traceback()
>   File "/home/airflow/.local/lib/python3.6/site-packages/azure/core/exceptions.py", line 218, in raise_with_traceback
>     raise super(AzureError, self).with_traceback(self.exc_traceback)
>   File "/home/airflow/.local/lib/python3.6/site-packages/azure/storage/blob/_upload_helpers.py", line 105, in upload_block_blob
>     **kwargs)
>   File "/home/airflow/.local/lib/python3.6/site-packages/azure/storage/blob/_generated/operations/_block_blob_operations.py", line 231, in upload
>     map_error(status_code=response.status_code, response=response, error_map=error_map)
>   File "/home/airflow/.local/lib/python3.6/site-packages/azure/core/exceptions.py", line 102, in map_error
>     raise error
> azure.core.exceptions.ResourceExistsError: The specified blob already exists.
> RequestId:8e0b61a7-c01e-0035-699d-4b837e000000
> Time:2021-05-18T04:19:41.7062904Z
> ErrorCode:BlobAlreadyExists
> Error:None
> `

What you expected to happen:

overwrite log file on backfills

How to reproduce it:

Run a trigger a dag run with remote logging wasb, delete dag and run again the same dag run.