task: metadata: install: pip install -e .[requests,aiohttp]
patch: diff --git a/google/resumable_media/requests/download.py b/google/resumable_media/requests/download.py
index 5d85b20..1472c9f 100644
--- a/google/resumable_media/requests/download.py
+++ b/google/resumable_media/requests/download.py
@@ -221,7 +221,7 @@ class Download(_request_helpers.RequestsMixin, _download.Download):
             # received using a range request, and set object generation query param.
             if self._bytes_downloaded > 0:
                 _download.add_bytes_range(
-                    self._bytes_downloaded, self.end, self._headers
+                    (self.start or 0) + self._bytes_downloaded, self.end, self._headers
                 )
                 request_kwargs["headers"] = self._headers
 
@@ -426,7 +426,7 @@ class RawDownload(_request_helpers.RawRequestsMixin, _download.Download):
             # received using a range request, and set object generation query param.
             if self._bytes_downloaded > 0:
                 _download.add_bytes_range(
-                    self._bytes_downloaded, self.end, self._headers
+                    (self.start or 0) + self._bytes_downloaded, self.end, self._headers
                 )
                 request_kwargs["headers"] = self._headers
 

problem_statement: Incorrect start offset when range reads are interrupted
#### Environment details

 GCP A100 machines
(unclear about exact software, as this bug was countered internally on Google infra when running on XManager Cloud)

#### Steps to reproduce

Given a range read of `start` to `end` bytes, if the read is interrupted, the code [here](https://github.com/googleapis/google-resumable-media-python/blob/5d2091d78fb71e0d63254c5b731afb05af432104/google/resumable_media/requests/download.py#L429) will make another GET request at the wrong start offset, as it should be `self.start + self._bytes_downloaded` if `self.start is not None`. Currently, the downloader will return `end + 1` bytes instead of `end - start + 1` bytes after resuming from interruption even when start is specified. 

The same issue also probably happens [here](https://github.com/googleapis/google-resumable-media-python/blob/5d2091d78fb71e0d63254c5b731afb05af432104/google/resumable_media/requests/download.py#L223) in `Downloader`.

We encountered this issue when using the [google python storage](https://github.com/googleapis/python-storage) library in the setting where we had thousands of workers doing repeated range reads on the same GCS file. Code would run and return the wrong bytes somewhere 20-60min in. Adding in a fix (see pull request) seemed to resolve the issue.

repo: googleapis/google-resumable-media-python
test_cmd: pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
test_patch: diff --git a/tests/unit/requests/test_download.py b/tests/unit/requests/test_download.py
index 91a50ec..713543c 100644
--- a/tests/unit/requests/test_download.py
+++ b/tests/unit/requests/test_download.py
@@ -455,6 +455,42 @@ class TestDownload(object):
         range_bytes = "bytes={:d}-{:d}".format(offset, end)
         assert download._headers["range"] == range_bytes
 
+    def test_consume_w_bytes_downloaded_range_read(self):
+        stream = io.BytesIO()
+        chunks = (b"up down ", b"charlie ", b"brown")
+        start = 1024
+        end = 65536
+
+        download = download_mod.Download(
+            EXAMPLE_URL,
+            stream=stream,
+            start=start,
+            end=end,
+            headers=None,
+            checksum="md5",
+        )
+        transport = mock.Mock(spec=["request"])
+        transport.request.return_value = _mock_response(chunks=chunks, headers=None)
+
+        assert download._bytes_downloaded == 0
+
+        # Mock a retry operation with bytes already downloaded in the stream and checksum stored
+        offset = 256
+        download._bytes_downloaded = offset
+        download._expected_checksum = None
+        download._checksum_object = _helpers._DoNothingHash()
+        download.consume(transport)
+
+        called_kwargs = {
+            "data": None,
+            "headers": download._headers,
+            "timeout": EXPECTED_TIMEOUT,
+            "stream": True,
+        }
+        transport.request.assert_called_once_with("GET", EXAMPLE_URL, **called_kwargs)
+        range_bytes = "bytes={:d}-{:d}".format(offset + start, end)
+        assert download._headers["range"] == range_bytes
+
     def test_consume_gzip_reset_stream_w_bytes_downloaded(self):
         stream = io.BytesIO()
         chunks = (b"up down ", b"charlie ", b"brown")
@@ -904,6 +940,42 @@ class TestRawDownload(object):
         range_bytes = "bytes={:d}-{:d}".format(offset, end)
         assert download._headers["range"] == range_bytes
 
+    def test_consume_w_bytes_downloaded_range_read(self):
+        stream = io.BytesIO()
+        chunks = (b"up down ", b"charlie ", b"brown")
+        start = 1024
+        end = 65536
+
+        download = download_mod.RawDownload(
+            EXAMPLE_URL,
+            stream=stream,
+            start=start,
+            end=end,
+            headers=None,
+            checksum="md5",
+        )
+        transport = mock.Mock(spec=["request"])
+        transport.request.return_value = _mock_raw_response(chunks=chunks, headers=None)
+
+        assert download._bytes_downloaded == 0
+
+        # Mock a retry operation with bytes already downloaded in the stream and checksum stored
+        offset = 256
+        download._bytes_downloaded = offset
+        download._expected_checksum = None
+        download._checksum_object = _helpers._DoNothingHash()
+        download.consume(transport)
+
+        called_kwargs = {
+            "data": None,
+            "headers": download._headers,
+            "timeout": EXPECTED_TIMEOUT,
+            "stream": True,
+        }
+        transport.request.assert_called_once_with("GET", EXAMPLE_URL, **called_kwargs)
+        range_bytes = "bytes={:d}-{:d}".format(start + offset, end)
+        assert download._headers["range"] == range_bytes
+
     def test_consume_gzip_reset_stream_w_bytes_downloaded(self):
         stream = io.BytesIO()
         chunks = (b"up down ", b"charlie ", b"brown")