diff --git a/storage/google/cloud/storage/blob.py b/storage/google/cloud/storage/blob.py
index 9c89c52b9e24..f2f582641c84 100644
--- a/storage/google/cloud/storage/blob.py
+++ b/storage/google/cloud/storage/blob.py
@@ -517,13 +517,17 @@ def _get_transport(self, client):
         client = self._require_client(client)
         return client._http
 
-    def _get_download_url(self):
+    def _get_download_url(self, query_params=None):
         """Get the download URL for the current blob.
 
         If the ``media_link`` has been loaded, it will be used, otherwise
         the URL will be constructed from the current blob's path (and possibly
         generation) to avoid a round trip.
 
+        :type query_params: dict or ``NoneType``
+        :param query_params: (Optional) Extension (custom) query parameters.
+                             See: https://cloud.google.com/storage/docs/json_api/v1/parameters
+
         :rtype: str
         :returns: The download URL for the current blob.
         """
@@ -538,6 +542,10 @@ def _get_download_url(self):
         if self.user_project is not None:
             name_value_pairs.append(("userProject", self.user_project))
 
+        if query_params is not None:
+            query_params_list = list(query_params.items())
+            name_value_pairs.extend(query_params_list)
+
         return _add_query_parameters(base_url, name_value_pairs)
 
     def _do_download(
@@ -586,7 +594,9 @@ def _do_download(
             while not download.finished:
                 download.consume_next_chunk(transport)
 
-    def download_to_file(self, file_obj, client=None, start=None, end=None):
+    def download_to_file(
+        self, file_obj, client=None, start=None, end=None, query_params=None
+    ):
         """Download the contents of this blob into a file-like object.
 
         .. note::
@@ -626,9 +636,13 @@ def download_to_file(self, file_obj, client=None, start=None, end=None):
         :type end: int
         :param end: Optional, The last byte in a range to be downloaded.
 
+        :type query_params: dict or ``NoneType``
+        :param query_params: (Optional) Extension (custom) query parameters.
+                             See: https://cloud.google.com/storage/docs/json_api/v1/parameters
+
         :raises: :class:`google.cloud.exceptions.NotFound`
         """
-        download_url = self._get_download_url()
+        download_url = self._get_download_url(query_params)
         headers = _get_encryption_headers(self._encryption_key)
         headers["accept-encoding"] = "gzip"
 
@@ -638,7 +652,9 @@ def download_to_file(self, file_obj, client=None, start=None, end=None):
         except resumable_media.InvalidResponse as exc:
             _raise_from_invalid_response(exc)
 
-    def download_to_filename(self, filename, client=None, start=None, end=None):
+    def download_to_filename(
+        self, filename, client=None, start=None, end=None, query_params=None
+    ):
         """Download the contents of this blob into a named file.
 
         If :attr:`user_project` is set on the bucket, bills the API request
@@ -658,11 +674,21 @@ def download_to_filename(self, filename, client=None, start=None, end=None):
         :type end: int
         :param end: Optional, The last byte in a range to be downloaded.
 
+        :type query_params: dict or ``NoneType``
+        :param query_params: (Optional) Extension (custom) query parameters.
+                             See: https://cloud.google.com/storage/docs/json_api/v1/parameters
+
         :raises: :class:`google.cloud.exceptions.NotFound`
         """
         try:
             with open(filename, "wb") as file_obj:
-                self.download_to_file(file_obj, client=client, start=start, end=end)
+                self.download_to_file(
+                    file_obj,
+                    client=client,
+                    start=start,
+                    end=end,
+                    query_params=query_params,
+                )
         except resumable_media.DataCorruption:
             # Delete the corrupt downloaded file.
             os.remove(filename)
@@ -673,7 +699,7 @@ def download_to_filename(self, filename, client=None, start=None, end=None):
             mtime = time.mktime(updated.timetuple())
             os.utime(file_obj.name, (mtime, mtime))
 
-    def download_as_string(self, client=None, start=None, end=None):
+    def download_as_string(self, client=None, start=None, end=None, query_params=None):
         """Download the contents of this blob as a string.
 
         If :attr:`user_project` is set on the bucket, bills the API request
@@ -690,12 +716,22 @@ def download_as_string(self, client=None, start=None, end=None):
         :type end: int
         :param end: Optional, The last byte in a range to be downloaded.
 
+        :type query_params: dict or ``NoneType``
+        :param query_params: (Optional) Extension (custom) query parameters.
+                             See: https://cloud.google.com/storage/docs/json_api/v1/parameters
+
         :rtype: bytes
         :returns: The data stored in this blob.
         :raises: :class:`google.cloud.exceptions.NotFound`
         """
         string_buffer = BytesIO()
-        self.download_to_file(string_buffer, client=client, start=start, end=end)
+        self.download_to_file(
+            string_buffer,
+            client=client,
+            start=start,
+            end=end,
+            query_params=query_params,
+        )
         return string_buffer.getvalue()
 
     def _get_content_type(self, content_type, filename=None):
@@ -784,7 +820,14 @@ def _get_upload_arguments(self, content_type):
         return headers, object_metadata, content_type
 
     def _do_multipart_upload(
-        self, client, stream, content_type, size, num_retries, predefined_acl
+        self,
+        client,
+        stream,
+        content_type,
+        size,
+        num_retries,
+        predefined_acl,
+        extra_headers,
     ):
         """Perform a multipart upload.
 
@@ -817,6 +860,10 @@ def _do_multipart_upload(
         :type predefined_acl: str
         :param predefined_acl: (Optional) predefined access control list
 
+        :type extra_headers: dict or ``NoneType``
+        :param extra_headers: (Optional) Extension (custom) query parameters.
+                              See: https://cloud.google.com/storage/docs/json_api/v1/parameters
+
         :rtype: :class:`~requests.Response`
         :returns: The "200 OK" response object returned after the multipart
                   upload request.
@@ -834,6 +881,8 @@ def _do_multipart_upload(
         transport = self._get_transport(client)
         info = self._get_upload_arguments(content_type)
         headers, object_metadata, content_type = info
+        if extra_headers is not None:
+            headers.update(extra_headers)
 
         base_url = _MULTIPART_URL_TEMPLATE.format(bucket_path=self.bucket.path)
         name_value_pairs = []
@@ -963,7 +1012,14 @@ def _initiate_resumable_upload(
         return upload, transport
 
     def _do_resumable_upload(
-        self, client, stream, content_type, size, num_retries, predefined_acl
+        self,
+        client,
+        stream,
+        content_type,
+        size,
+        num_retries,
+        predefined_acl,
+        extra_headers,
     ):
         """Perform a resumable upload.
 
@@ -998,6 +1054,10 @@ def _do_resumable_upload(
         :type predefined_acl: str
         :param predefined_acl: (Optional) predefined access control list
 
+        :type extra_headers: dict or ``NoneType``
+        :param extra_headers: (Optional) Extension (custom) query parameters.
+                              See: https://cloud.google.com/storage/docs/json_api/v1/parameters
+
         :rtype: :class:`~requests.Response`
         :returns: The "200 OK" response object returned after the final chunk
                   is uploaded.
@@ -1009,6 +1069,7 @@ def _do_resumable_upload(
             size,
             num_retries,
             predefined_acl=predefined_acl,
+            extra_headers=extra_headers,
         )
 
         while not upload.finished:
@@ -1017,7 +1078,14 @@ def _do_resumable_upload(
         return response
 
     def _do_upload(
-        self, client, stream, content_type, size, num_retries, predefined_acl
+        self,
+        client,
+        stream,
+        content_type,
+        size,
+        num_retries,
+        predefined_acl,
+        extra_headers,
     ):
         """Determine an upload strategy and then perform the upload.
 
@@ -1054,6 +1122,10 @@ def _do_upload(
         :type predefined_acl: str
         :param predefined_acl: (Optional) predefined access control list
 
+        :type extra_headers: dict or ``NoneType``
+        :param extra_headers: (Optional) Extension (custom) query parameters.
+                              See: https://cloud.google.com/storage/docs/json_api/v1/parameters
+
         :rtype: dict
         :returns: The parsed JSON from the "200 OK" response. This will be the
                   **only** response in the multipart case and it will be the
@@ -1061,11 +1133,23 @@ def _do_upload(
         """
         if size is not None and size <= _MAX_MULTIPART_SIZE:
             response = self._do_multipart_upload(
-                client, stream, content_type, size, num_retries, predefined_acl
+                client,
+                stream,
+                content_type,
+                size,
+                num_retries,
+                predefined_acl,
+                extra_headers,
             )
         else:
             response = self._do_resumable_upload(
-                client, stream, content_type, size, num_retries, predefined_acl
+                client,
+                stream,
+                content_type,
+                size,
+                num_retries,
+                predefined_acl,
+                extra_headers,
             )
 
         return response.json()
@@ -1079,6 +1163,7 @@ def upload_from_file(
         num_retries=None,
         client=None,
         predefined_acl=None,
+        extra_headers=None,
     ):
         """Upload the contents of this blob from a file-like object.
 
@@ -1140,6 +1225,10 @@ def upload_from_file(
         :type predefined_acl: str
         :param predefined_acl: (Optional) predefined access control list
 
+        :type extra_headers: dict or ``NoneType``
+        :param extra_headers: (Optional) Extension (custom) query parameters.
+                              See: https://cloud.google.com/storage/docs/json_api/v1/parameters
+
         :raises: :class:`~google.cloud.exceptions.GoogleCloudError`
                  if the upload response returns an error status.
 
@@ -1155,14 +1244,25 @@ def upload_from_file(
 
         try:
             created_json = self._do_upload(
-                client, file_obj, content_type, size, num_retries, predefined_acl
+                client,
+                file_obj,
+                content_type,
+                size,
+                num_retries,
+                predefined_acl,
+                extra_headers,
             )
             self._set_properties(created_json)
         except resumable_media.InvalidResponse as exc:
             _raise_from_invalid_response(exc)
 
     def upload_from_filename(
-        self, filename, content_type=None, client=None, predefined_acl=None
+        self,
+        filename,
+        content_type=None,
+        client=None,
+        predefined_acl=None,
+        extra_headers=None,
     ):
         """Upload this blob's contents from the content of a named file.
 
@@ -1200,6 +1300,10 @@ def upload_from_filename(
 
         :type predefined_acl: str
         :param predefined_acl: (Optional) predefined access control list
+
+        :type extra_headers: dict or ``NoneType``
+        :param extra_headers: (Optional) Extension (custom) query parameters.
+                              See: https://cloud.google.com/storage/docs/json_api/v1/parameters
         """
         content_type = self._get_content_type(content_type, filename=filename)
 
@@ -1211,10 +1315,16 @@ def upload_from_filename(
                 client=client,
                 size=total_bytes,
                 predefined_acl=predefined_acl,
+                extra_headers=extra_headers,
             )
 
     def upload_from_string(
-        self, data, content_type="text/plain", client=None, predefined_acl=None
+        self,
+        data,
+        content_type="text/plain",
+        client=None,
+        predefined_acl=None,
+        extra_headers=None,
     ):
         """Upload contents of this blob from the provided string.
 
@@ -1247,6 +1357,10 @@ def upload_from_string(
 
         :type predefined_acl: str
         :param predefined_acl: (Optional) predefined access control list
+
+        :type extra_headers: dict or ``NoneType``
+        :param extra_headers: (Optional) Extension (custom) query parameters.
+                              See: https://cloud.google.com/storage/docs/json_api/v1/parameters
         """
         data = _to_bytes(data, encoding="utf-8")
         string_buffer = BytesIO(data)
@@ -1256,10 +1370,11 @@ def upload_from_string(
             content_type=content_type,
             client=client,
             predefined_acl=predefined_acl,
+            extra_headers=extra_headers,
         )
 
     def create_resumable_upload_session(
-        self, content_type=None, size=None, origin=None, client=None
+        self, content_type=None, size=None, origin=None, client=None, extra_headers={}
     ):
         """Create a resumable upload session.
 
@@ -1322,10 +1437,13 @@ def create_resumable_upload_session(
                   completed by making an HTTP PUT request with the
                   file's contents.
 
+        :type extra_headers: dict
+        :param extra_headers: (Optional) Extension (custom) query parameters.
+                              See: https://cloud.google.com/storage/docs/json_api/v1/parameters
+
         :raises: :class:`google.cloud.exceptions.GoogleCloudError`
                  if the session creation response returns an error status.
         """
-        extra_headers = {}
         if origin is not None:
             # This header is specifically for client-side uploads, it
             # determines the origins allowed for CORS.
diff --git a/storage/google/cloud/storage/client.py b/storage/google/cloud/storage/client.py
index a3aa5d628fa9..d3634380e5c8 100644
--- a/storage/google/cloud/storage/client.py
+++ b/storage/google/cloud/storage/client.py
@@ -353,7 +353,9 @@ def create_bucket(self, bucket_or_name, requester_pays=None, project=None):
         bucket.create(client=self, project=project)
         return bucket
 
-    def download_blob_to_file(self, blob_or_uri, file_obj, start=None, end=None):
+    def download_blob_to_file(
+        self, blob_or_uri, file_obj, start=None, end=None, query_params=None
+    ):
         """Download the contents of a blob object or blob URI into a file-like object.
 
         Args:
@@ -368,6 +370,9 @@ def download_blob_to_file(self, blob_or_uri, file_obj, start=None, end=None):
                 Optional. The first byte in a range to be downloaded.
             end (int):
                 Optional. The last byte in a range to be downloaded.
+            query_params (dict):
+                (Optional) Extension (Custom) query parameters configurations.
+                See: https://cloud.google.com/storage/docs/json_api/v1/parameters
 
         Examples:
             Download a blob using using a blob resource.
@@ -394,7 +399,9 @@ def download_blob_to_file(self, blob_or_uri, file_obj, start=None, end=None):
 
         """
         try:
-            blob_or_uri.download_to_file(file_obj, client=self, start=start, end=end)
+            blob_or_uri.download_to_file(
+                file_obj, client=self, start=start, end=end, query_params=query_params
+            )
         except AttributeError:
             scheme, netloc, path, query, frag = urlsplit(blob_or_uri)
             if scheme != "gs":
@@ -402,7 +409,9 @@ def download_blob_to_file(self, blob_or_uri, file_obj, start=None, end=None):
             bucket = Bucket(self, name=netloc)
             blob_or_uri = Blob(path[1:], bucket)
 
-            blob_or_uri.download_to_file(file_obj, client=self, start=start, end=end)
+            blob_or_uri.download_to_file(
+                file_obj, client=self, start=start, end=end, query_params=query_params
+            )
 
     def list_blobs(
         self,
@@ -475,7 +484,7 @@ def list_blobs(
             versions=versions,
             projection=projection,
             fields=fields,
-            client=self
+            client=self,
         )
 
     def list_buckets(
diff --git a/storage/tests/unit/test_blob.py b/storage/tests/unit/test_blob.py
index b264ddcb8acf..04ec234bc532 100644
--- a/storage/tests/unit/test_blob.py
+++ b/storage/tests/unit/test_blob.py
@@ -348,9 +348,7 @@ def test_public_url_w_tilde_in_name(self):
         BLOB_NAME = "foo~bar"
         bucket = _Bucket()
         blob = self._make_one(BLOB_NAME, bucket=bucket)
-        self.assertEqual(
-            blob.public_url, "https://storage.googleapis.com/name/foo~bar"
-        )
+        self.assertEqual(blob.public_url, "https://storage.googleapis.com/name/foo~bar")
 
     def test_public_url_with_non_ascii(self):
         blob_name = u"winter \N{snowman}"
@@ -683,6 +681,18 @@ def test__get_download_url_with_media_link_w_user_project(self):
             download_url, "{}?userProject={}".format(media_link, user_project)
         )
 
+    def test__get_download_url_with_media_link_w_query_params(self):
+        blob_name = "something.txt"
+        bucket = _Bucket(name="IRRELEVANT")
+        blob = self._make_one(blob_name, bucket=bucket)
+        media_link = "http://test.invalid"
+        # Set the media link on the blob
+        blob._properties["mediaLink"] = media_link
+        query_params = {"prettyPrint": False}
+        download_url = blob._get_download_url(query_params=query_params)
+        result = "{}?prettyPrint={}".format(media_link, query_params["prettyPrint"])
+        self.assertEqual(download_url, result)
+
     def test__get_download_url_on_the_fly(self):
         blob_name = "bzzz-fly.txt"
         bucket = _Bucket(name="buhkit")
@@ -974,7 +984,7 @@ def test_download_to_file_wo_media_link(self):
         )
         self._check_session_mocks(client, transport, expected_url)
 
-    def _download_to_file_helper(self, use_chunks=False):
+    def _download_to_file_helper(self, use_chunks=False, query_params=None):
         blob_name = "blob-name"
         transport = self._mock_download_transport()
         # Create a fake client/bucket and use them in the Blob() constructor.
@@ -998,9 +1008,12 @@ def _download_to_file_helper(self, use_chunks=False):
             transport.request.side_effect = [single_chunk_response]
 
         file_obj = io.BytesIO()
-        blob.download_to_file(file_obj)
+        blob.download_to_file(file_obj, query_params=query_params)
         self.assertEqual(file_obj.getvalue(), b"abcdef")
-
+        if query_params is not None:
+            media_link = "{}?prettyPrint={}".format(
+                media_link, query_params["prettyPrint"]
+            )
         if use_chunks:
             self._check_session_mocks(client, transport, media_link)
         else:
@@ -1018,6 +1031,9 @@ def test_download_to_file_default(self):
     def test_download_to_file_with_chunk_size(self):
         self._download_to_file_helper(use_chunks=True)
 
+    def test_download_to_file_with_query_param(self):
+        self._download_to_file_helper(query_params={"prettyPrint": False})
+
     def _download_to_filename_helper(self, updated=None):
         import os
         import time
@@ -1292,6 +1308,7 @@ def _do_multipart_success(
         user_project=None,
         predefined_acl=None,
         kms_key_name=None,
+        extra_headers=None,
     ):
         from six.moves.urllib.parse import urlencode
 
@@ -1308,7 +1325,13 @@ def _do_multipart_success(
         stream = io.BytesIO(data)
         content_type = u"application/xml"
         response = blob._do_multipart_upload(
-            client, stream, content_type, size, num_retries, predefined_acl
+            client,
+            stream,
+            content_type,
+            size,
+            num_retries,
+            predefined_acl,
+            extra_headers,
         )
 
         # Check the mocks and the returned value.
@@ -1347,6 +1370,8 @@ def _do_multipart_success(
             + b"\r\n--==0==--"
         )
         headers = {"content-type": b'multipart/related; boundary="==0=="'}
+        if extra_headers is not None:
+            headers.update(extra_headers)
         transport.request.assert_called_once_with(
             "POST", upload_url, data=payload, headers=headers
         )
@@ -1387,12 +1412,18 @@ def test__do_multipart_upload_bad_size(self):
         self.assertGreater(size, len(data))
 
         with self.assertRaises(ValueError) as exc_info:
-            blob._do_multipart_upload(None, stream, None, size, None, None)
+            blob._do_multipart_upload(None, stream, None, size, None, None, None)
 
         exc_contents = str(exc_info.exception)
         self.assertIn("was specified but the file-like object only had", exc_contents)
         self.assertEqual(stream.tell(), len(data))
 
+    @mock.patch(u"google.resumable_media._upload.get_boundary", return_value=b"==0==")
+    def test__do_multipart_upload_with_headers(self, mock_get_boundary):
+        self._do_multipart_success(
+            mock_get_boundary, extra_headers={"X-Goog-Testing": "true"}
+        )
+
     def _initiate_resumable_helper(
         self,
         size=None,
@@ -1565,7 +1596,9 @@ def _make_resumable_transport(self, headers1, headers2, headers3, total_bytes):
         return fake_transport, responses
 
     @staticmethod
-    def _do_resumable_upload_call0(blob, content_type, size=None, predefined_acl=None):
+    def _do_resumable_upload_call0(
+        blob, content_type, size=None, predefined_acl=None, extra_headers=None
+    ):
         # First mock transport.request() does initiates upload.
         upload_url = (
             "https://www.googleapis.com/upload/storage/v1"
@@ -1580,6 +1613,8 @@ def _do_resumable_upload_call0(blob, content_type, size=None, predefined_acl=Non
         }
         if size is not None:
             expected_headers["x-upload-content-length"] = str(size)
+        if extra_headers is not None:
+            expected_headers.update(extra_headers)
         payload = json.dumps({"name": blob.name}).encode("utf-8")
         return mock.call("POST", upload_url, data=payload, headers=expected_headers)
 
@@ -1616,7 +1651,7 @@ def _do_resumable_upload_call2(
         return mock.call("PUT", resumable_url, data=payload, headers=expected_headers)
 
     def _do_resumable_helper(
-        self, use_size=False, num_retries=None, predefined_acl=None
+        self, use_size=False, num_retries=None, predefined_acl=None, extra_headers=None
     ):
         bucket = _Bucket(name="yesterday")
         blob = self._make_one(u"blob-name", bucket=bucket)
@@ -1644,7 +1679,13 @@ def _do_resumable_helper(
         stream = io.BytesIO(data)
         content_type = u"text/html"
         response = blob._do_resumable_upload(
-            client, stream, content_type, size, num_retries, predefined_acl
+            client,
+            stream,
+            content_type,
+            size,
+            num_retries,
+            predefined_acl,
+            extra_headers,
         )
 
         # Check the returned values.
@@ -1653,7 +1694,11 @@ def _do_resumable_helper(
 
         # Check the mocks.
         call0 = self._do_resumable_upload_call0(
-            blob, content_type, size=size, predefined_acl=predefined_acl
+            blob,
+            content_type,
+            size=size,
+            predefined_acl=predefined_acl,
+            extra_headers=extra_headers,
         )
         call1 = self._do_resumable_upload_call1(
             blob,
@@ -1685,8 +1730,16 @@ def test__do_resumable_upload_with_retry(self):
     def test__do_resumable_upload_with_predefined_acl(self):
         self._do_resumable_helper(predefined_acl="private")
 
+    def test__do_resumable_upload_with_header(self):
+        self._do_resumable_helper(extra_headers={"X-Goog-Testing": "true"})
+
     def _do_upload_helper(
-        self, chunk_size=None, num_retries=None, predefined_acl=None, size=None
+        self,
+        chunk_size=None,
+        num_retries=None,
+        predefined_acl=None,
+        size=None,
+        extra_headers=None,
     ):
         blob = self._make_one(u"blob-name", bucket=None)
 
@@ -1710,19 +1763,37 @@ def _do_upload_helper(
             size = 12345654321
         # Make the request and check the mocks.
         created_json = blob._do_upload(
-            client, stream, content_type, size, num_retries, predefined_acl
+            client,
+            stream,
+            content_type,
+            size,
+            num_retries,
+            predefined_acl,
+            extra_headers,
         )
         self.assertIs(created_json, mock.sentinel.json)
         response.json.assert_called_once_with()
         if size is not None and size <= google.cloud.storage.blob._MAX_MULTIPART_SIZE:
             blob._do_multipart_upload.assert_called_once_with(
-                client, stream, content_type, size, num_retries, predefined_acl
+                client,
+                stream,
+                content_type,
+                size,
+                num_retries,
+                predefined_acl,
+                extra_headers,
             )
             blob._do_resumable_upload.assert_not_called()
         else:
             blob._do_multipart_upload.assert_not_called()
             blob._do_resumable_upload.assert_called_once_with(
-                client, stream, content_type, size, num_retries, predefined_acl
+                client,
+                stream,
+                content_type,
+                size,
+                num_retries,
+                predefined_acl,
+                extra_headers,
             )
 
     def test__do_upload_uses_multipart(self):
@@ -1734,6 +1805,9 @@ def test__do_upload_uses_resumable(self):
             size=google.cloud.storage.blob._MAX_MULTIPART_SIZE + 1,
         )
 
+    def test__do_upload_uses_extra_header(self):
+        self._do_upload_helper(extra_headers={"X-Goog-Testing": "true"})
+
     def test__do_upload_with_retry(self):
         self._do_upload_helper(num_retries=20)
 
@@ -1755,6 +1829,7 @@ def _upload_from_file_helper(self, side_effect=None, **kwargs):
         content_type = u"font/woff"
         client = mock.sentinel.client
         predefined_acl = kwargs.get("predefined_acl", None)
+        extra_headers = kwargs.get("extra_headers", None)
         ret_val = blob.upload_from_file(
             stream, size=len(data), content_type=content_type, client=client, **kwargs
         )
@@ -1767,12 +1842,20 @@ def _upload_from_file_helper(self, side_effect=None, **kwargs):
         # Check the mock.
         num_retries = kwargs.get("num_retries")
         blob._do_upload.assert_called_once_with(
-            client, stream, content_type, len(data), num_retries, predefined_acl
+            client,
+            stream,
+            content_type,
+            len(data),
+            num_retries,
+            predefined_acl,
+            extra_headers,
         )
         return stream
 
     def test_upload_from_file_success(self):
-        stream = self._upload_from_file_helper(predefined_acl="private")
+        stream = self._upload_from_file_helper(
+            predefined_acl="private", extra_headers={"X-Goog-Testing": "true"}
+        )
         assert stream.tell() == 2
 
     @mock.patch("warnings.warn")
@@ -1785,7 +1868,7 @@ def test_upload_from_file_with_retries(self, mock_warn):
         )
 
     def test_upload_from_file_with_rewind(self):
-        stream = self._upload_from_file_helper(rewind=True)
+        stream = self._upload_from_file_helper(rewind=True, extra_headers=None)
         assert stream.tell() == 0
 
     def test_upload_from_file_failure(self):
@@ -1801,7 +1884,7 @@ def test_upload_from_file_failure(self):
         side_effect = InvalidResponse(response, message)
 
         with self.assertRaises(exceptions.Conflict) as exc_info:
-            self._upload_from_file_helper(side_effect=side_effect)
+            self._upload_from_file_helper(side_effect=side_effect, extra_headers={})
 
         self.assertIn(message, exc_info.exception.message)
         self.assertEqual(exc_info.exception.errors, [])
@@ -1811,12 +1894,13 @@ def _do_upload_mock_call_helper(self, blob, client, content_type, size):
         mock_call = blob._do_upload.mock_calls[0]
         call_name, pos_args, kwargs = mock_call
         self.assertEqual(call_name, "")
-        self.assertEqual(len(pos_args), 6)
+        self.assertEqual(len(pos_args), 7)
         self.assertEqual(pos_args[0], client)
         self.assertEqual(pos_args[2], content_type)
         self.assertEqual(pos_args[3], size)
         self.assertIsNone(pos_args[4])  # num_retries
         self.assertIsNone(pos_args[5])  # predefined_acl
+        self.assertIsNone(pos_args[6])  # extra_headers
         self.assertEqual(kwargs, {})
 
         return pos_args[1]
diff --git a/storage/tests/unit/test_client.py b/storage/tests/unit/test_client.py
index 89ed8570dbfa..e90b5433d2e0 100644
--- a/storage/tests/unit/test_client.py
+++ b/storage/tests/unit/test_client.py
@@ -619,7 +619,7 @@ def test_download_blob_to_file_with_blob(self):
 
         client.download_blob_to_file(blob, file_obj)
         blob.download_to_file.assert_called_once_with(
-            file_obj, client=client, start=None, end=None
+            file_obj, client=client, start=None, end=None, query_params=None
         )
 
     def test_download_blob_to_file_with_uri(self):
@@ -633,7 +633,7 @@ def test_download_blob_to_file_with_uri(self):
             client.download_blob_to_file("gs://bucket_name/path/to/object", file_obj)
 
         blob.download_to_file.assert_called_once_with(
-            file_obj, client=client, start=None, end=None
+            file_obj, client=client, start=None, end=None, query_params=None
         )
 
     def test_download_blob_to_file_with_invalid_uri(self):
@@ -651,6 +651,7 @@ def test_download_blob_to_file_with_invalid_uri(self):
 
     def test_list_blobs(self):
         from google.cloud.storage.bucket import Bucket
+
         BUCKET_NAME = "bucket-name"
 
         credentials = _make_credentials()
@@ -658,8 +659,8 @@ def test_list_blobs(self):
         connection = _make_connection({"items": []})
 
         with mock.patch(
-            'google.cloud.storage.client.Client._connection',
-            new_callable=mock.PropertyMock
+            "google.cloud.storage.client.Client._connection",
+            new_callable=mock.PropertyMock,
         ) as client_mock:
             client_mock.return_value = connection
 
@@ -671,11 +672,12 @@ def test_list_blobs(self):
             connection.api_request.assert_called_once_with(
                 method="GET",
                 path="/b/%s/o" % BUCKET_NAME,
-                query_params={"projection": "noAcl"}
+                query_params={"projection": "noAcl"},
             )
 
     def test_list_blobs_w_all_arguments_and_user_project(self):
         from google.cloud.storage.bucket import Bucket
+
         BUCKET_NAME = "name"
         USER_PROJECT = "user-project-123"
         MAX_RESULTS = 10
@@ -701,8 +703,8 @@ def test_list_blobs_w_all_arguments_and_user_project(self):
         connection = _make_connection({"items": []})
 
         with mock.patch(
-            'google.cloud.storage.client.Client._connection',
-            new_callable=mock.PropertyMock
+            "google.cloud.storage.client.Client._connection",
+            new_callable=mock.PropertyMock,
         ) as client_mock:
             client_mock.return_value = connection
 
@@ -721,9 +723,7 @@ def test_list_blobs_w_all_arguments_and_user_project(self):
 
             self.assertEqual(blobs, [])
             connection.api_request.assert_called_once_with(
-                method="GET",
-                path="/b/%s/o" % BUCKET_NAME,
-                query_params=EXPECTED
+                method="GET", path="/b/%s/o" % BUCKET_NAME, query_params=EXPECTED
             )
 
     def test_list_buckets_wo_project(self):