diff --git a/google/cloud/storage/batch.py b/google/cloud/storage/batch.py
index 732439f14..a60df80f8 100644
--- a/google/cloud/storage/batch.py
+++ b/google/cloud/storage/batch.py
@@ -276,6 +276,11 @@ def finish(self):
         response = self._client._base_connection._make_request(
             "POST", url, data=body, headers=headers, timeout=timeout
         )
+
+        # Raise exception if the top-level batch request fails
+        if not 200 <= response.status_code < 300:
+            raise exceptions.from_http_response(response)
+
         responses = list(_unpack_batch_response(response))
         self._finish_futures(responses)
         return responses
diff --git a/tests/unit/test_batch.py b/tests/unit/test_batch.py
index d43f27e8e..f877448f8 100644
--- a/tests/unit/test_batch.py
+++ b/tests/unit/test_batch.py
@@ -447,6 +447,24 @@ def test_finish_nonempty_non_multipart_response(self):
         with self.assertRaises(ValueError):
             batch.finish()
 
+    def test_finish_multipart_response_with_status_failure(self):
+        from google.cloud.exceptions import ServiceUnavailable
+
+        url = "http://api.example.com/other_api"
+        expected_response = _make_response(
+            status=http_client.SERVICE_UNAVAILABLE,
+            headers={"content-type": 'multipart/mixed; boundary="DEADBEEF="'},
+        )
+        http = _make_requests_session([expected_response])
+        connection = _Connection(http=http)
+        client = _Client(connection)
+        batch = self._make_one(client)
+        batch.API_BASE_URL = "http://api.example.com"
+        batch._requests.append(("POST", url, {}, {"foo": 1, "bar": 2}, None))
+
+        with self.assertRaises(ServiceUnavailable):
+            batch.finish()
+
     def test_as_context_mgr_wo_error(self):
         from google.cloud.storage.client import Client