Register the S3 request signer on the session

pyiceberg/io/fsspec.py

@@ -219,13 +219,19 @@ def _s3(properties: Properties) -> AbstractFileSystem:
     if profile_name := get_first_property_value(properties, S3_PROFILE_NAME, AWS_PROFILE_NAME):
         s3_fs_kwargs["profile"] = profile_name
 
-    fs = S3FileSystem(**s3_fs_kwargs)
-
-    for event_name, event_function in register_events.items():
-        fs.s3.meta.events.unregister(event_name, unique_id=1925)
-        fs.s3.meta.events.register_last(event_name, event_function, unique_id=1925)
-
-    return fs
+    if register_events:
+        from aiobotocore.session import AioSession
+
+        # Register on the session rather than on an already-constructed client. S3FileSystem
+        # creates the client that issues requests lazily inside the running event loop, and that
+        # client does not inherit handlers attached to an earlier one.
+        session = AioSession()
+        event_emitter = session.get_component("event_emitter")
+        for event_name, event_function in register_events.items():
+            event_emitter.register_last(event_name, event_function, unique_id=1925)
+        s3_fs_kwargs["session"] = session
+
+    return S3FileSystem(**s3_fs_kwargs)
 
 
 def _gs(properties: Properties) -> AbstractFileSystem:

tests/io/test_fsspec.py

@@ -15,6 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
+import asyncio
 import os
 import pickle
 import tempfile
@@ -23,6 +24,7 @@
 from unittest import mock
 
 import pytest
+from botocore import UNSIGNED
 from botocore.awsrequest import AWSRequest
 from fsspec.implementations.local import LocalFileSystem
 from fsspec.spec import AbstractFileSystem
@@ -877,6 +879,51 @@ def _test_fsspec_pickle_round_trip(fsspec_fileio: FsspecFileIO, location: str) -
 TEST_URI = "https://iceberg-test-signer"
 
 
+def test_s3v4_rest_signer_registered_on_session(requests_mock: Mocker) -> None:
+    """Signer must be registered on the session, which every client the filesystem creates inherits."""
+    new_uri = "https://other-bucket/metadata/snap-8048355899640248710-1-a5c8ea2d-aa1f-48e8-89f4-1fa69db8c742.avro"
+    requests_mock.post(
+        f"{TEST_URI}/v1/aws/s3/sign",
+        json={
+            "uri": new_uri,
+            "headers": {"Authorization": ["AWS4-HMAC-SHA256 Credential=ASIAQPRZZYGHUT57DL3I/20221017/us-west-2/s3/aws4_request"]},
+            "extensions": {},
+        },
+        status_code=200,
+    )
+
+    with mock.patch("s3fs.S3FileSystem") as mock_s3fs:
+        s3_fileio = FsspecFileIO(properties={"s3.signer": "S3V4RestSigner", "uri": TEST_URI, "token": "abc"})
+        s3_fileio.new_input(location="s3://warehouse/foo")
+
+    session = mock_s3fs.call_args.kwargs["session"]
+
+    request = AWSRequest(
+        method="HEAD",
+        url="https://bucket/metadata/snap-8048355899640248710-1-a5c8ea2d-aa1f-48e8-89f4-1fa69db8c742.avro",
+        headers={},
+        data=b"",
+        params={},
+        auth_path="/metadata/snap-8048355899640248710-1-a5c8ea2d-aa1f-48e8-89f4-1fa69db8c742.avro",
+    )
+    request.context = {"client_region": "us-west-2"}
+
+    asyncio.run(
+        session.get_component("event_emitter").emit(
+            "before-sign.s3",
+            request=request,
+            signing_name="s3",
+            region_name="us-west-2",
+            signature_version=UNSIGNED,
+            request_signer=None,
+            operation_name="HeadObject",
+        )
+    )
+
+    assert request.url == new_uri
+    assert dict(request.headers)["Authorization"].startswith("AWS4-HMAC-SHA256")
+
+
 def test_s3v4_rest_signer(requests_mock: Mocker) -> None:
     new_uri = "https://other-bucket/metadata/snap-8048355899640248710-1-a5c8ea2d-aa1f-48e8-89f4-1fa69db8c742.avro"
     requests_mock.post(