fix(storage): re-open gRPC stream on finalize retry attempt

Author: sahusneha2004

packages/google-cloud-storage/google/cloud/storage/asyncio/async_appendable_object_writer.py

@@ -681,7 +681,19 @@ async def finalize(
         if retry_policy is None:
             retry_policy = AsyncRetry(predicate=_is_write_retryable)
 
+        attempt_count = 0
+
         async def _do_finalize():
+            nonlocal attempt_count
+            attempt_count += 1
+
+            if attempt_count > 1:
+                logger.info(
+                    f"Re-opening the stream for finalize retry attempt: {attempt_count}"
+                )
+                self._is_stream_open = False
+                await self.open()
+
             await self.write_obj_stream.send(finalize_req)
             response = await self.write_obj_stream.recv()
             self.object_resource = response.resource
@@ -691,7 +703,19 @@ async def _do_finalize():
         try:
             return await retry_policy(_do_finalize)()
         finally:
-            await self.write_obj_stream.close()
+            if self.write_obj_stream:
+                if self.write_obj_stream.is_stream_open:
+                    try:
+                        await self.write_obj_stream.close()
+                    except Exception as e:
+                        logger.warning(
+                            f"Error closing write stream during finalize cleanup. Got exception: {e}"
+                        )
+                else:
+                    try:
+                        await self.write_obj_stream.close()
+                    except Exception:
+                        pass
             self._is_stream_open = False
             self.offset = None