<p dir="auto"><strong>Apache Airflow version</strong>:<br>
1.10.10</p>
<p dir="auto"><strong>Kubernetes version (if you are using kubernetes)</strong> (use <code class="notranslate">kubectl version</code>):<br>
v1.16.15-gke.6000</p>
<p dir="auto"><strong>Environment</strong>:</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>: GCP</li>
<li><strong>OS</strong> (e.g. from /etc/os-release): Debian GNU/Linux 10 (buster)</li>
<li><strong>Kernel</strong> (e.g. <code class="notranslate">uname -a</code>): 4.19.112+</li>
<li><strong>Install tools</strong>:</li>
<li><strong>Others</strong>:</li>
</ul>
<p dir="auto"><strong>What happened</strong>:<br>
I get duplicated data when run task by <code class="notranslate">GoogleCloudStorageToBigQueryOperator</code>.<br>
The task log has error message: Connection reset by peer when running job, task failed and retry is fine, but problem is <strong>the executing BQ job doesn’t be cancel, so previous job and retry job both append same data to destination.</strong></p>
<p dir="auto"><strong>What you expected to happen</strong>:<br>
The previous running job must be cancel when task failed.</p>

<p dir="auto"><strong>How to reproduce it</strong>: Network issue only happens at sometimes.</p>

<p dir="auto"><strong>Anything else we need to know</strong>:<br>
Once per week.</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="[2021-02-16 00:46:58,584] {{discovery.py:867}} INFO - URL being requested: GET https://bigquery.googleapis.com/bigquery/v2/projects/dcard-production/jobs/job_CmfL_czbPNebWFysuoCGYz6I6b31?location=US&amp;alt=json
[2021-02-16 00:46:58,587] {{taskinstance.py:1150}} ERROR - [Errno 104] Connection reset by peer
Traceback (most recent call last):
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py&quot;, line 984, in _run_raw_task
    result = task_copy.execute(context=context)
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/contrib/operators/gcs_to_bq.py&quot;, line 288, in execute
    encryption_configuration=self.encryption_configuration)
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/contrib/hooks/bigquery_hook.py&quot;, line 1302, in run_load
    return self.run_with_configuration(configuration)
  File &quot;/usr/local/lib/python3.7/site-packages/airflow/contrib/hooks/bigquery_hook.py&quot;, line 1338, in run_with_configuration
    location=location).execute(num_retries=self.num_retries)
  File &quot;/usr/local/lib/python3.7/site-packages/googleapiclient/_helpers.py&quot;, line 130, in positional_wrapper
    return wrapped(*args, **kwargs)
  File &quot;/usr/local/lib/python3.7/site-packages/googleapiclient/http.py&quot;, line 851, in execute
    method=str(self.method), body=self.body, headers=self.headers)
  File &quot;/usr/local/lib/python3.7/site-packages/googleapiclient/http.py&quot;, line 165, in _retry_request
    resp, content = http.request(uri, method, *args, **kwargs)
  File &quot;/usr/local/lib/python3.7/site-packages/google_auth_httplib2.py&quot;, line 198, in request
    uri, method, body=body, headers=request_headers, **kwargs)
  File &quot;/usr/local/lib/python3.7/site-packages/googleapiclient/http.py&quot;, line 1729, in new_request
    redirections=redirections, connection_type=connection_type)
  File &quot;/usr/local/lib/python3.7/site-packages/httplib2/__init__.py&quot;, line 1957, in request
    cachekey,
  File &quot;/usr/local/lib/python3.7/site-packages/httplib2/__init__.py&quot;, line 1622, in _request
    conn, request_uri, method, body, headers
  File &quot;/usr/local/lib/python3.7/site-packages/httplib2/__init__.py&quot;, line 1560, in _conn_request
    response = conn.getresponse()
  File &quot;/usr/local/lib/python3.7/http/client.py&quot;, line 1369, in getresponse
    response.begin()
  File &quot;/usr/local/lib/python3.7/http/client.py&quot;, line 310, in begin
    version, status, reason = self._read_status()
  File &quot;/usr/local/lib/python3.7/http/client.py&quot;, line 271, in _read_status
    line = str(self.fp.readline(_MAXLINE + 1), &quot;iso-8859-1&quot;)
  File &quot;/usr/local/lib/python3.7/socket.py&quot;, line 589, in readinto
    return self._sock.recv_into(b)
  File &quot;/usr/local/lib/python3.7/ssl.py&quot;, line 1071, in recv_into
    return self.read(nbytes, buffer)
  File &quot;/usr/local/lib/python3.7/ssl.py&quot;, line 929, in read
    return self._sslobj.read(len, buffer)
ConnectionResetError: [Errno 104] Connection reset by peer
[2021-02-16 00:46:58,595] {{taskinstance.py:1194}} INFO - Marking task as UP_FOR_RETRY. dag_id=core_data_postgres_hourly_v6, task_id=truncate_or_append_dcard_post_collections_with_new_records, execution_date=20210215T230000, start_date=20210216T004651, end_date=20210216T004658
[2021-02-16 00:47:01,011] {{local_task_job.py:102}} INFO - Task exited with return code 1"><pre class="notranslate"><code class="notranslate">[2021-02-16 00:46:58,584] {{discovery.py:867}} INFO - URL being requested: GET https://bigquery.googleapis.com/bigquery/v2/projects/dcard-production/jobs/job_CmfL_czbPNebWFysuoCGYz6I6b31?location=US&amp;alt=json
[2021-02-16 00:46:58,587] {{taskinstance.py:1150}} ERROR - [Errno 104] Connection reset by peer
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 984, in _run_raw_task
    result = task_copy.execute(context=context)
  File "/usr/local/lib/python3.7/site-packages/airflow/contrib/operators/gcs_to_bq.py", line 288, in execute
    encryption_configuration=self.encryption_configuration)
  File "/usr/local/lib/python3.7/site-packages/airflow/contrib/hooks/bigquery_hook.py", line 1302, in run_load
    return self.run_with_configuration(configuration)
  File "/usr/local/lib/python3.7/site-packages/airflow/contrib/hooks/bigquery_hook.py", line 1338, in run_with_configuration
    location=location).execute(num_retries=self.num_retries)
  File "/usr/local/lib/python3.7/site-packages/googleapiclient/_helpers.py", line 130, in positional_wrapper
    return wrapped(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/googleapiclient/http.py", line 851, in execute
    method=str(self.method), body=self.body, headers=self.headers)
  File "/usr/local/lib/python3.7/site-packages/googleapiclient/http.py", line 165, in _retry_request
    resp, content = http.request(uri, method, *args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/google_auth_httplib2.py", line 198, in request
    uri, method, body=body, headers=request_headers, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/googleapiclient/http.py", line 1729, in new_request
    redirections=redirections, connection_type=connection_type)
  File "/usr/local/lib/python3.7/site-packages/httplib2/__init__.py", line 1957, in request
    cachekey,
  File "/usr/local/lib/python3.7/site-packages/httplib2/__init__.py", line 1622, in _request
    conn, request_uri, method, body, headers
  File "/usr/local/lib/python3.7/site-packages/httplib2/__init__.py", line 1560, in _conn_request
    response = conn.getresponse()
  File "/usr/local/lib/python3.7/http/client.py", line 1369, in getresponse
    response.begin()
  File "/usr/local/lib/python3.7/http/client.py", line 310, in begin
    version, status, reason = self._read_status()
  File "/usr/local/lib/python3.7/http/client.py", line 271, in _read_status
    line = str(self.fp.readline(_MAXLINE + 1), "iso-8859-1")
  File "/usr/local/lib/python3.7/socket.py", line 589, in readinto
    return self._sock.recv_into(b)
  File "/usr/local/lib/python3.7/ssl.py", line 1071, in recv_into
    return self.read(nbytes, buffer)
  File "/usr/local/lib/python3.7/ssl.py", line 929, in read
    return self._sslobj.read(len, buffer)
ConnectionResetError: [Errno 104] Connection reset by peer
[2021-02-16 00:46:58,595] {{taskinstance.py:1194}} INFO - Marking task as UP_FOR_RETRY. dag_id=core_data_postgres_hourly_v6, task_id=truncate_or_append_dcard_post_collections_with_new_records, execution_date=20210215T230000, start_date=20210216T004651, end_date=20210216T004658
[2021-02-16 00:47:01,011] {{local_task_job.py:102}} INFO - Task exited with return code 1
</code></pre></div>
<p dir="auto">Hello,</p>
<p dir="auto">We use Python 3+, so we should consider using an enumerated type for objects that are buckets for contents  This will give us better type hints.<br>
<a href="https://github.com/apache/airflow/blob/master/airflow/utils/state.py">https://github.com/apache/airflow/blob/master/airflow/utils/state.py</a></p>
<p dir="auto">Best regards,<br>
Kamil Breguła</p>