diff --git a/java-storage/google-cloud-storage/pom.xml b/java-storage/google-cloud-storage/pom.xml
index 892fd729b9ae..5fcfb43904aa 100644
--- a/java-storage/google-cloud-storage/pom.xml
+++ b/java-storage/google-cloud-storage/pom.xml
@@ -296,7 +296,6 @@
     <dependency>
       <groupId>com.google.api.grpc</groupId>
       <artifactId>proto-google-cloud-storage-control-v2</artifactId>
-      <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/AcoSpan.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/AcoSpan.java
new file mode 100644
index 000000000000..13cfeb8fe2bb
--- /dev/null
+++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/AcoSpan.java
@@ -0,0 +1,169 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.api.trace.StatusCode;
+import java.util.concurrent.TimeUnit;
+
+final class AcoSpan implements Span {
+  private final Span delegate;
+  private final String bucketName;
+  private final OtelStorageDecorator parent;
+
+  AcoSpan(Span delegate, String bucketName, OtelStorageDecorator parent) {
+    this.delegate = delegate;
+    this.bucketName = bucketName;
+    this.parent = parent;
+  }
+
+  private void applyCacheAttributes() {
+    if (bucketName != null && parent != null && parent.delegate instanceof StorageInternal) {
+      BucketMetadataCache.BucketMetadata md =
+          ((StorageInternal) parent.delegate).getBucketMetadataCache().get(bucketName);
+      if (md != null) {
+        delegate.setAttribute("gcp.resource.destination.id", md.resource);
+        delegate.setAttribute("gcp.resource.destination.location", md.location);
+      }
+    }
+  }
+
+  @Override
+  public void end() {
+    applyCacheAttributes();
+    delegate.end();
+  }
+
+  @Override
+  public void end(long timestamp, TimeUnit unit) {
+    applyCacheAttributes();
+    delegate.end(timestamp, unit);
+  }
+
+  @Override
+  public Span recordException(Throwable exception) {
+    delegate.recordException(exception);
+    if (exception instanceof StorageException
+        && parent != null
+        && parent.delegate instanceof StorageInternal) {
+      StorageException se = (StorageException) exception;
+      if (se.getCode() == 404) {
+        ((StorageInternal) parent.delegate).getBucketMetadataCache().remove(bucketName);
+      }
+    }
+    return this;
+  }
+
+  @Override
+  public Span recordException(Throwable exception, Attributes attributes) {
+    delegate.recordException(exception, attributes);
+    if (exception instanceof StorageException
+        && parent != null
+        && parent.delegate instanceof StorageInternal) {
+      StorageException se = (StorageException) exception;
+      if (se.getCode() == 404) {
+        ((StorageInternal) parent.delegate).getBucketMetadataCache().remove(bucketName);
+      }
+    }
+    return this;
+  }
+
+  @Override
+  public Span setAttribute(String k, String v) {
+    delegate.setAttribute(k, v);
+    return this;
+  }
+
+  @Override
+  public Span setAttribute(String k, long v) {
+    delegate.setAttribute(k, v);
+    return this;
+  }
+
+  @Override
+  public Span setAttribute(String k, double v) {
+    delegate.setAttribute(k, v);
+    return this;
+  }
+
+  @Override
+  public Span setAttribute(String k, boolean v) {
+    delegate.setAttribute(k, v);
+    return this;
+  }
+
+  @Override
+  public <T> Span setAttribute(AttributeKey<T> k, T v) {
+    delegate.setAttribute(k, v);
+    return this;
+  }
+
+  @Override
+  public Span addEvent(String n) {
+    delegate.addEvent(n);
+    return this;
+  }
+
+  @Override
+  public Span addEvent(String n, Attributes a) {
+    delegate.addEvent(n, a);
+    return this;
+  }
+
+  @Override
+  public Span addEvent(String n, long t, TimeUnit u) {
+    delegate.addEvent(n, t, u);
+    return this;
+  }
+
+  @Override
+  public Span addEvent(String n, Attributes a, long t, TimeUnit u) {
+    delegate.addEvent(n, a, t, u);
+    return this;
+  }
+
+  @Override
+  public Span setStatus(StatusCode c) {
+    delegate.setStatus(c);
+    return this;
+  }
+
+  @Override
+  public Span setStatus(StatusCode c, String d) {
+    delegate.setStatus(c, d);
+    return this;
+  }
+
+  @Override
+  public Span updateName(String name) {
+    delegate.updateName(name);
+    return this;
+  }
+
+  @Override
+  public SpanContext getSpanContext() {
+    return delegate.getSpanContext();
+  }
+
+  @Override
+  public boolean isRecording() {
+    return delegate.isRecording();
+  }
+}
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/AcoSpanBuilder.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/AcoSpanBuilder.java
new file mode 100644
index 000000000000..bfcc2a1f77c1
--- /dev/null
+++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/AcoSpanBuilder.java
@@ -0,0 +1,130 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanBuilder;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.context.Context;
+import java.util.concurrent.TimeUnit;
+
+final class AcoSpanBuilder implements SpanBuilder {
+  private final SpanBuilder delegate;
+  private final OtelStorageDecorator parent;
+  private String bucketName;
+
+  AcoSpanBuilder(SpanBuilder delegate, OtelStorageDecorator parent) {
+    this.delegate = delegate;
+    this.parent = parent;
+  }
+
+  @Override
+  public SpanBuilder setAttribute(String key, String value) {
+    delegate.setAttribute(key, value);
+    if ("gsutil.uri".equals(key) && value != null) {
+      String name = OtelStorageDecorator.extractBucketName(value);
+      if (name != null && !name.isEmpty()) {
+        this.bucketName = name;
+      }
+    }
+    return this;
+  }
+
+  @Override
+  public <T> SpanBuilder setAttribute(AttributeKey<T> key, T value) {
+    delegate.setAttribute(key, value);
+    if ("gsutil.uri".equals(key.getKey()) && value instanceof String) {
+      String name = OtelStorageDecorator.extractBucketName((String) value);
+      if (name != null && !name.isEmpty()) {
+        this.bucketName = name;
+      }
+    }
+    return this;
+  }
+
+  @Override
+  public Span startSpan() {
+    if (bucketName != null && parent != null && parent.delegate instanceof StorageInternal) {
+      parent.checkCacheAndTriggerFetch(bucketName);
+      BucketMetadataCache.BucketMetadata md =
+          ((StorageInternal) parent.delegate).getBucketMetadataCache().get(bucketName);
+      if (md != null) {
+        delegate.setAttribute("gcp.resource.destination.id", md.resource);
+        delegate.setAttribute("gcp.resource.destination.location", md.location);
+      }
+      return new AcoSpan(delegate.startSpan(), bucketName, parent);
+    }
+    return delegate.startSpan();
+  }
+
+  @Override
+  public SpanBuilder setNoParent() {
+    delegate.setNoParent();
+    return this;
+  }
+
+  @Override
+  public SpanBuilder setAttribute(String key, boolean value) {
+    delegate.setAttribute(key, value);
+    return this;
+  }
+
+  @Override
+  public SpanBuilder setAttribute(String key, double value) {
+    delegate.setAttribute(key, value);
+    return this;
+  }
+
+  @Override
+  public SpanBuilder setAttribute(String key, long value) {
+    delegate.setAttribute(key, value);
+    return this;
+  }
+
+  @Override
+  public SpanBuilder setSpanKind(SpanKind k) {
+    delegate.setSpanKind(k);
+    return this;
+  }
+
+  @Override
+  public SpanBuilder setStartTimestamp(long t, TimeUnit u) {
+    delegate.setStartTimestamp(t, u);
+    return this;
+  }
+
+  @Override
+  public SpanBuilder setParent(Context c) {
+    delegate.setParent(c);
+    return this;
+  }
+
+  @Override
+  public SpanBuilder addLink(SpanContext c) {
+    delegate.addLink(c);
+    return this;
+  }
+
+  @Override
+  public SpanBuilder addLink(SpanContext c, Attributes a) {
+    delegate.addLink(c, a);
+    return this;
+  }
+}
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BucketMetadataCache.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BucketMetadataCache.java
new file mode 100644
index 000000000000..ef87b6601451
--- /dev/null
+++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BucketMetadataCache.java
@@ -0,0 +1,76 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+final class BucketMetadataCache {
+
+  static final class BucketMetadata {
+    final String resource;
+    final String location;
+
+    BucketMetadata(String resource, String location) {
+      this.resource = resource;
+      this.location = location;
+    }
+  }
+
+  private final Object lock = new Object();
+  private final Map<String, BucketMetadata> cache;
+
+  BucketMetadataCache(int capacity) {
+    this.cache =
+        new LinkedHashMap<String, BucketMetadata>(16, 0.75f, true) {
+          @Override
+          protected boolean removeEldestEntry(Map.Entry<String, BucketMetadata> eldest) {
+            return size() > capacity;
+          }
+        };
+  }
+
+  BucketMetadata get(String bucketName) {
+    synchronized (lock) {
+      return cache.get(bucketName);
+    }
+  }
+
+  void put(String bucketName, BucketMetadata metadata) {
+    synchronized (lock) {
+      cache.put(bucketName, metadata);
+    }
+  }
+
+  void remove(String bucketName) {
+    synchronized (lock) {
+      cache.remove(bucketName);
+    }
+  }
+
+  void clear() {
+    synchronized (lock) {
+      cache.clear();
+    }
+  }
+
+  boolean containsKey(String bucketName) {
+    synchronized (lock) {
+      return cache.containsKey(bucketName);
+    }
+  }
+}
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
index 120b7a269724..acc8b28b9973 100644
--- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
+++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageImpl.java
@@ -82,6 +82,9 @@
 import com.google.iam.v1.GetIamPolicyRequest;
 import com.google.iam.v1.SetIamPolicyRequest;
 import com.google.iam.v1.TestIamPermissionsRequest;
+import com.google.storage.control.v2.GetStorageLayoutRequest;
+import com.google.storage.control.v2.StorageLayout;
+import com.google.storage.control.v2.StorageLayoutName;
 import com.google.storage.v2.AppendObjectSpec;
 import com.google.storage.v2.BidiReadObjectRequest;
 import com.google.storage.v2.BidiReadObjectSpec;
@@ -112,6 +115,8 @@
 import com.google.storage.v2.WriteObjectRequest;
 import com.google.storage.v2.WriteObjectResponse;
 import com.google.storage.v2.WriteObjectSpec;
+import io.grpc.protobuf.ProtoUtils;
+import io.grpc.stub.ClientCalls;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -180,6 +185,8 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
   // workaround for https://github.com/googleapis/java-storage/issues/1736
   private final Opts<UserProject> defaultOpts;
   @Deprecated private final Supplier<ProjectId> defaultProjectId;
+  private volatile BucketMetadataCache bucketMetadataCache;
+  private final java.lang.Object cacheInitLock = new java.lang.Object();
 
   GrpcStorageImpl(
       GrpcStorageOptions options,
@@ -202,6 +209,38 @@ final class GrpcStorageImpl extends BaseService<StorageOptions>
     this.defaultProjectId = Suppliers.memoize(() -> UnifiedOpts.projectId(options.getProjectId()));
   }
 
+  @Override
+  public BucketMetadataCache getBucketMetadataCache() {
+    if (bucketMetadataCache == null) {
+      synchronized (cacheInitLock) {
+        if (bucketMetadataCache == null) {
+          bucketMetadataCache = new BucketMetadataCache(10000);
+        }
+      }
+    }
+    return bucketMetadataCache;
+  }
+
+  @Override
+  public com.google.cloud.Tuple<String, String> internalGetStorageLayout(String bucketName) {
+    com.google.storage.v2.stub.StorageStub rawStub = storageClient.getStub();
+    if (!(rawStub instanceof GrpcStorageOptions.AcoGrpcStorageStub)) {
+      throw new RuntimeException("StorageStub is not an AcoGrpcStorageStub");
+    }
+    GrpcStorageOptions.AcoGrpcStorageStub stub = (GrpcStorageOptions.AcoGrpcStorageStub) rawStub;
+
+    GetStorageLayoutRequest request =
+        GetStorageLayoutRequest.newBuilder()
+            .setName(StorageLayoutName.of(getOptions().getProjectId(), bucketName).toString())
+            .build();
+
+    com.google.api.gax.grpc.GrpcCallContext merge = com.google.cloud.storage.Utils.merge(com.google.api.gax.grpc.GrpcCallContext.createDefault(), Retrying.newCallContext());
+
+    StorageLayout layout = stub.getStorageLayoutCallable().call(request, merge);
+
+    return com.google.cloud.Tuple.of(layout.getName(), layout.getLocation());
+  }
+
   @Override
   public void close() throws Exception {
     try (StorageClient s = storageClient;
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
index 1a6726b9c01b..478926b75cde 100644
--- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
+++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/GrpcStorageOptions.java
@@ -45,6 +45,10 @@
 import com.google.api.gax.rpc.RequestParamsBuilder;
 import com.google.api.gax.rpc.RequestParamsExtractor;
 import com.google.api.gax.rpc.ServerStreamingCallable;
+import com.google.api.gax.rpc.UnaryCallable;
+import com.google.api.gax.grpc.GrpcCallSettings;
+import com.google.storage.control.v2.GetStorageLayoutRequest;
+import com.google.storage.control.v2.StorageLayout;
 import com.google.api.gax.rpc.internal.QuotaProjectIdHidingCredentials;
 import com.google.api.gax.tracing.ApiTracerFactory;
 import com.google.api.pathtemplate.PathTemplate;
@@ -951,7 +955,10 @@ public Storage create(StorageOptions options) {
           } else {
             LOGGER.config(
                 "zero-copy protobuf deserialization unavailable, proceeding with default");
-            StorageClient client = StorageClient.create(storageSettings);
+            StorageStubSettings baseSettings = (StorageStubSettings) storageSettings.getStubSettings();
+            AcoStorageStubSettings.Builder acoStubBuilder = new AcoStorageStubSettings.Builder(baseSettings);
+            AcoStorageSettings.Builder acoSettingsBuilder = new AcoStorageSettings.Builder(acoStubBuilder);
+            AcoStorageClient client = new AcoStorageClient(new AcoStorageSettings(acoSettingsBuilder));
             StorageDataClient dataClient =
                 StorageDataClient.create(
                     executor,
@@ -1103,6 +1110,94 @@ public InternalStorageSettings build() throws IOException {
     }
   }
 
+  static class AcoGrpcStorageStub extends GrpcStorageStub {
+    private final UnaryCallable<GetStorageLayoutRequest, StorageLayout> getStorageLayoutCallable;
+
+    AcoGrpcStorageStub(
+        StorageStubSettings settings,
+        ClientContext clientContext,
+        GrpcStubCallableFactory callableFactory)
+        throws IOException {
+      super(settings, clientContext, callableFactory);
+
+      MethodDescriptor<GetStorageLayoutRequest, StorageLayout> getStorageLayoutMethod =
+          MethodDescriptor.<GetStorageLayoutRequest, StorageLayout>newBuilder()
+              .setType(MethodDescriptor.MethodType.UNARY)
+              .setFullMethodName("google.storage.control.v2.StorageControl/GetStorageLayout")
+              .setRequestMarshaller(ProtoUtils.marshaller(GetStorageLayoutRequest.getDefaultInstance()))
+              .setResponseMarshaller(ProtoUtils.marshaller(StorageLayout.getDefaultInstance()))
+              .build();
+
+      GrpcCallSettings<GetStorageLayoutRequest, StorageLayout> transportSettings =
+          GrpcCallSettings.<GetStorageLayoutRequest, StorageLayout>newBuilder()
+              .setMethodDescriptor(getStorageLayoutMethod)
+              .setParamsExtractor(request -> ImmutableMap.of())
+              .build();
+
+      this.getStorageLayoutCallable =
+          callableFactory.createUnaryCallable(
+              transportSettings,
+              com.google.api.gax.rpc.UnaryCallSettings.<GetStorageLayoutRequest, StorageLayout>newUnaryCallSettingsBuilder().build(),
+              clientContext);
+    }
+
+    public UnaryCallable<GetStorageLayoutRequest, StorageLayout> getStorageLayoutCallable() {
+      return getStorageLayoutCallable;
+    }
+  }
+
+  private static final class AcoStorageStubSettings extends StorageStubSettings {
+    private AcoStorageStubSettings(Builder settingsBuilder) throws IOException {
+      super(settingsBuilder);
+    }
+    @Override
+    public StorageStub createStub() throws IOException {
+      if (!getTransportChannelProvider()
+          .getTransportName()
+          .equals(GrpcTransportChannel.getGrpcTransportName())) {
+        throw new UnsupportedOperationException(
+            String.format(
+                "Transport not supported: %s", getTransportChannelProvider().getTransportName()));
+      }
+      ClientContext clientContext = ClientContext.create(this);
+      return new AcoGrpcStorageStub(this, clientContext, new GrpcStorageCallableFactory());
+    }
+    private static final class Builder extends StorageStubSettings.Builder {
+      private Builder(StorageStubSettings settings) {
+        super(settings);
+      }
+      @Override
+      public AcoStorageStubSettings build() throws IOException {
+        return new AcoStorageStubSettings(this);
+      }
+    }
+  }
+
+  private static final class AcoStorageSettings extends StorageSettings {
+    private AcoStorageSettings(Builder settingsBuilder) throws IOException {
+      super(settingsBuilder);
+    }
+    private static final class Builder extends StorageSettings.Builder {
+      private Builder(StorageStubSettings.Builder stubSettings) {
+        super(stubSettings);
+      }
+      @Override
+      public AcoStorageSettings build() throws IOException {
+        return new AcoStorageSettings(this);
+      }
+    }
+  }
+
+  private static final class AcoStorageClient extends StorageClient {
+    private AcoStorageClient(StorageSettings settings) throws IOException {
+      super(settings);
+    }
+    @Override
+    public AcoGrpcStorageStub getStub() {
+      return (AcoGrpcStorageStub) super.getStub();
+    }
+  }
+
   private static final class InternalStorageStubSettings extends StorageStubSettings {
 
     private InternalStorageStubSettings(Builder settingsBuilder) throws IOException {
@@ -1141,7 +1236,7 @@ public InternalStorageStubSettings build() throws IOException {
 
   // DanglingJavadocs are for breadcrumbs to source of copied generated code
   @SuppressWarnings("DanglingJavadoc")
-  private static final class InternalZeroCopyGrpcStorageStub extends GrpcStorageStub
+  private static final class InternalZeroCopyGrpcStorageStub extends AcoGrpcStorageStub
       implements AutoCloseable {
 
     private static final RequestParamsExtractor<BidiReadObjectRequest>
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelMultipartUploadClientDecorator.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelMultipartUploadClientDecorator.java
index f5e7080fed75..a3832b2e9529 100644
--- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelMultipartUploadClientDecorator.java
+++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelMultipartUploadClientDecorator.java
@@ -53,7 +53,7 @@ private OtelMultipartUploadClientDecorator(
     this.delegate = delegate;
     this.tracer =
         OtelStorageDecorator.TracerDecorator.decorate(
-            null, otel, baseAttributes, MultipartUploadClient.class.getName() + "/");
+            null, null, otel, baseAttributes, MultipartUploadClient.class.getName() + "/");
   }
 
   @Override
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java
index 291db00ae5d3..c0be8f617819 100644
--- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java
+++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/OtelStorageDecorator.java
@@ -58,6 +58,8 @@
 import java.nio.file.Path;
 import java.util.List;
 import java.util.Locale;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.function.UnaryOperator;
 import org.checkerframework.checker.nullness.qual.NonNull;
@@ -81,7 +83,54 @@ private OtelStorageDecorator(Storage delegate, OpenTelemetry otel, Attributes ba
     this.otel = otel;
     this.baseAttributes = baseAttributes;
     this.tracer =
-        TracerDecorator.decorate(null, otel, baseAttributes, Storage.class.getName() + "/");
+        TracerDecorator.decorate(this, null, otel, baseAttributes, Storage.class.getName() + "/");
+  }
+
+  static String extractBucketName(String uri) {
+    if (uri == null || !uri.startsWith("gs://")) {
+      return null;
+    }
+    String remainder = uri.substring(5);
+    int firstSlash = remainder.indexOf('/');
+    if (firstSlash == -1) {
+      return remainder;
+    }
+    return remainder.substring(0, firstSlash);
+  }
+
+  private final ExecutorService cacheExecutor =
+      Executors.newCachedThreadPool(
+          r -> {
+            Thread t = new Thread(r);
+            t.setDaemon(true);
+            t.setName("gcs-aco-metadata-cache-pool");
+            return t;
+          });
+
+  void checkCacheAndTriggerFetch(String bucketName) {
+    if (!(delegate instanceof StorageInternal)) {
+      return;
+    }
+    StorageInternal internal = (StorageInternal) delegate;
+    BucketMetadataCache cache = internal.getBucketMetadataCache();
+
+    if (cache.containsKey(bucketName)) {
+      return;
+    }
+
+    cache.put(
+        bucketName,
+        new BucketMetadataCache.BucketMetadata("projects/_/buckets/" + bucketName, "global"));
+
+    cacheExecutor.submit(
+        () -> {
+          try {
+            com.google.cloud.Tuple<String, String> layout =
+                internal.internalGetStorageLayout(bucketName);
+            cache.put(bucketName, new BucketMetadataCache.BucketMetadata(layout.x(), layout.y()));
+          } catch (Exception e) {
+          }
+        });
   }
 
   @Override
@@ -1423,7 +1472,14 @@ public boolean deleteNotification(String bucket, String notificationId) {
 
   @Override
   public void close() throws Exception {
-    delegate.close();
+    try {
+      if (delegate instanceof StorageInternal) {
+        ((StorageInternal) delegate).getBucketMetadataCache().clear();
+      }
+      cacheExecutor.shutdownNow();
+    } finally {
+      delegate.close();
+    }
   }
 
   @Override
@@ -1562,16 +1618,19 @@ static UnaryOperator<RetryContext> retryContextDecorator(OpenTelemetry otel) {
   }
 
   static final class TracerDecorator implements Tracer {
+    @Nullable private final OtelStorageDecorator parentDecorator;
     @Nullable private final Context parentContextOverride;
     private final Tracer delegate;
     private final Attributes baseAttributes;
     private final String spanNamePrefix;
 
     TracerDecorator(
+        @Nullable OtelStorageDecorator parentDecorator,
         @Nullable Context parentContextOverride,
         Tracer delegate,
         Attributes baseAttributes,
         String spanNamePrefix) {
+      this.parentDecorator = parentDecorator;
       this.parentContextOverride = parentContextOverride;
       this.delegate = delegate;
       this.baseAttributes = baseAttributes;
@@ -1579,6 +1638,7 @@ static final class TracerDecorator implements Tracer {
     }
 
     static TracerDecorator decorate(
+        @Nullable OtelStorageDecorator parentDecorator,
         @Nullable Context parentContextOverride,
         OpenTelemetry otel,
         Attributes baseAttributes,
@@ -1588,7 +1648,8 @@ static TracerDecorator decorate(
       requireNonNull(spanNamePrefix, "spanNamePrefix must be non null");
       Tracer tracer =
           otel.getTracer(OTEL_SCOPE_NAME, StorageOptions.getDefaultInstance().getLibraryVersion());
-      return new TracerDecorator(parentContextOverride, tracer, baseAttributes, spanNamePrefix);
+      return new TracerDecorator(
+          parentDecorator, parentContextOverride, tracer, baseAttributes, spanNamePrefix);
     }
 
     @Override
@@ -1598,7 +1659,8 @@ public SpanBuilder spanBuilder(String spanName) {
       if (parentContextOverride != null) {
         spanBuilder.setParent(parentContextOverride);
       }
-      return spanBuilder;
+
+      return new AcoSpanBuilder(spanBuilder, parentDecorator);
     }
   }
 
@@ -1671,6 +1733,7 @@ public OtelDecoratedBlobWriteSession(BlobWriteSession delegate, Span sessionSpan
       this.sessionSpan = sessionSpan;
       this.tracer =
           TracerDecorator.decorate(
+              OtelStorageDecorator.this,
               Context.current(),
               otel,
               OtelStorageDecorator.this.baseAttributes,
@@ -1794,6 +1857,7 @@ public OtelDecoratedCopyWriter(CopyWriter copyWriter, Span span) {
       this.parentContext = Context.current();
       this.tracer =
           TracerDecorator.decorate(
+              OtelStorageDecorator.this,
               Context.current(),
               otel,
               OtelStorageDecorator.this.baseAttributes,
@@ -2127,6 +2191,7 @@ private OtelDecoratingBlobAppendableUpload(BlobAppendableUpload delegate, Span u
       this.uploadSpan = uploadSpan;
       this.tracer =
           TracerDecorator.decorate(
+              OtelStorageDecorator.this,
               Context.current(),
               otel,
               OtelStorageDecorator.this.baseAttributes,
@@ -2163,6 +2228,7 @@ private OtelDecoratingAppendableUploadWriteableByteChannel(
         this.openSpan = openSpan;
         this.tracer =
             TracerDecorator.decorate(
+                OtelStorageDecorator.this,
                 Context.current(),
                 otel,
                 OtelStorageDecorator.this.baseAttributes,
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
index 8a510b84e7cc..0f1037b05caf 100644
--- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
+++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageImpl.java
@@ -117,6 +117,25 @@ final class StorageImpl extends BaseService<StorageOptions> implements Storage,
   final StorageRpc storageRpc;
   final WriterFactory writerFactory;
   final Retrier retrier;
+  private volatile BucketMetadataCache bucketMetadataCache;
+  private final java.lang.Object cacheInitLock = new java.lang.Object();
+
+  @Override
+  public BucketMetadataCache getBucketMetadataCache() {
+    if (bucketMetadataCache == null) {
+      synchronized (cacheInitLock) {
+        if (bucketMetadataCache == null) {
+          bucketMetadataCache = new BucketMetadataCache(10000);
+        }
+      }
+    }
+    return bucketMetadataCache;
+  }
+
+  @Override
+  public com.google.cloud.Tuple<String, String> internalGetStorageLayout(String bucketName) {
+    return storageRpc.getStorageLayout(bucketName);
+  }
 
   StorageImpl(HttpStorageOptions options, WriterFactory writerFactory, Retrier retrier) {
     super(options);
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java
index 0d700c46df24..e92179eb1596 100644
--- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java
+++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/StorageInternal.java
@@ -48,4 +48,12 @@ default BlobInfo compose(ComposeRequest composeRequest) {
   default BlobInfo internalObjectGet(BlobId blobId, Opts<ObjectSourceOpt> opts) {
     throw new UnsupportedOperationException("not implemented");
   }
+
+  default com.google.cloud.Tuple<String, String> internalGetStorageLayout(String bucketName) {
+    throw new UnsupportedOperationException("not implemented");
+  }
+
+  default BucketMetadataCache getBucketMetadataCache() {
+    throw new UnsupportedOperationException("not implemented");
+  }
 }
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
index 97814b597c37..5db559914c8f 100644
--- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
+++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java
@@ -583,6 +583,37 @@ public Bucket get(Bucket bucket, Map<Option, ?> options) {
     }
   }
 
+  @Override
+  public com.google.cloud.Tuple<String, String> getStorageLayout(String bucketName) {
+    try {
+      String url = options.getHost() + "/storage/v1/b/" + bucketName + "/storageLayout";
+      com.google.api.client.http.GenericUrl genericUrl =
+          new com.google.api.client.http.GenericUrl(url);
+      com.google.api.client.http.HttpRequest request =
+          storage.getRequestFactory().buildGetRequest(genericUrl);
+      com.google.api.client.http.HttpResponse response = request.execute();
+      String content = response.parseAsString();
+
+      String actualResource = "projects/_/buckets/" + bucketName;
+      String actualLocation = "global";
+
+      com.google.api.client.json.JsonParser parser =
+          storage.getJsonFactory().createJsonParser(content);
+      @SuppressWarnings("unchecked")
+      Map<String, Object> map = parser.parse(Map.class);
+      if (map.containsKey("name")) {
+        actualResource = (String) map.get("name");
+      }
+      if (map.containsKey("location")) {
+        actualLocation = (String) map.get("location");
+      }
+
+      return com.google.cloud.Tuple.of(actualResource, actualLocation);
+    } catch (IOException e) {
+      throw translate(e);
+    }
+  }
+
   private Storage.Objects.Get getCall(StorageObject object, Map<Option, ?> options)
       throws IOException {
     Storage.Objects.Get get = storage.objects().get(object.getBucket(), object.getName());
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java
index 33f24a54c854..636c8a147792 100644
--- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java
+++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/StorageRpc.java
@@ -640,6 +640,13 @@ TestIamPermissionsResponse testIamPermissions(
    */
   ServiceAccount getServiceAccount(String projectId);
 
+  /**
+   * Returns the storage layout for the specified bucket.
+   *
+   * @throws StorageException upon failure
+   */
+  com.google.cloud.Tuple<String, String> getStorageLayout(String bucketName);
+
   @InternalApi
   Storage getStorage();
 }
diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/testing/StorageRpcTestBase.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/testing/StorageRpcTestBase.java
index 8f835f5bf3f2..2140e13110e8 100644
--- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/testing/StorageRpcTestBase.java
+++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/testing/StorageRpcTestBase.java
@@ -322,6 +322,11 @@ public ServiceAccount getServiceAccount(String projectId) {
     throw new UnsupportedOperationException("Not implemented yet");
   }
 
+  @Override
+  public com.google.cloud.Tuple<String, String> getStorageLayout(String bucketName) {
+    throw new UnsupportedOperationException("Not implemented yet");
+  }
+
   @Override
   public StorageObject moveObject(
       String bucket,
diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITOpenTelemetryTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITOpenTelemetryTest.java
index 3b8957bbac64..90fe2b6cdd8f 100644
--- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITOpenTelemetryTest.java
+++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITOpenTelemetryTest.java
@@ -63,20 +63,32 @@ public void checkInstrumentation() throws Exception {
         storage.getOptions().toBuilder().setOpenTelemetry(openTelemetrySdk).build();
     try (Storage storage = storageOptions.getService()) {
       storage.create(BlobInfo.newBuilder(bucket, generator.randomObjectName()).build());
+      Thread.sleep(800);
+      storage.create(BlobInfo.newBuilder(bucket, generator.randomObjectName()).build());
     }
 
-    SpanData spanData = exporter.getExportedSpans().get(0);
+    assertThat(exporter.getExportedSpans().size()).isAtLeast(2);
+    SpanData span1 = exporter.getExportedSpans().get(0);
+    SpanData span2 = exporter.getExportedSpans().get(1);
+
     assertAll(
-        () -> assertThat(getAttributeValue(spanData, "gcp.client.service")).isEqualTo("Storage"),
+        () -> assertThat(getAttributeValue(span1, "gcp.client.service")).isEqualTo("Storage"),
+        () ->
+            assertThat(getAttributeValue(span1, "rpc.system"))
+                .isEqualTo(transport.name().toLowerCase()),
+        () ->
+            assertThat(getAttributeValue(span1, "gcp.resource.destination.id"))
+                .isEqualTo("projects/_/buckets/" + bucket.getName()),
         () ->
-            assertThat(getAttributeValue(spanData, "gcp.client.repo"))
-                .isEqualTo("googleapis/java-storage"),
+            assertThat(getAttributeValue(span1, "gcp.resource.destination.location"))
+                .isEqualTo("global"),
+        () -> assertThat(getAttributeValue(span2, "gcp.client.service")).isEqualTo("Storage"),
         () ->
-            assertThat(getAttributeValue(spanData, "gcp.client.artifact"))
-                .isEqualTo("com.google.cloud:google-cloud-storage"),
+            assertThat(getAttributeValue(span2, "gcp.resource.destination.id"))
+                .contains("buckets/" + bucket.getName()),
         () ->
-            assertThat(getAttributeValue(spanData, "rpc.system"))
-                .isEqualTo(transport.name().toLowerCase()));
+            assertThat(getAttributeValue(span2, "gcp.resource.destination.location"))
+                .isNotEqualTo("global"));
   }
 
   @Test
diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/OtelStorageDecoratorAcoUnitTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/OtelStorageDecoratorAcoUnitTest.java
new file mode 100644
index 000000000000..c2e27234bd1d
--- /dev/null
+++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/OtelStorageDecoratorAcoUnitTest.java
@@ -0,0 +1,66 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.storage;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.mockito.Mockito.mock;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanBuilder;
+import io.opentelemetry.api.trace.Tracer;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+public class OtelStorageDecoratorAcoUnitTest {
+
+  @Test
+  public void testCacheAndProxyDecorationFlow() throws Exception {
+    Storage mockStorage =
+        mock(Storage.class, Mockito.withSettings().extraInterfaces(StorageInternal.class));
+    StorageInternal mockInternal = (StorageInternal) mockStorage;
+
+    BucketMetadataCache cache = new BucketMetadataCache(10000);
+    Mockito.when(mockInternal.getBucketMetadataCache()).thenReturn(cache);
+
+    OpenTelemetry mockOtel = mock(OpenTelemetry.class);
+    Tracer mockTracer = mock(Tracer.class);
+    SpanBuilder mockSpanBuilder = mock(SpanBuilder.class);
+    Span mockSpan = mock(Span.class);
+
+    Mockito.when(mockOtel.getTracer(Mockito.anyString(), Mockito.anyString()))
+        .thenReturn(mockTracer);
+    Mockito.when(mockTracer.spanBuilder(Mockito.anyString())).thenReturn(mockSpanBuilder);
+    Mockito.when(mockSpanBuilder.setAllAttributes(Mockito.any())).thenReturn(mockSpanBuilder);
+    Mockito.when(mockSpanBuilder.startSpan()).thenReturn(mockSpan);
+
+    Storage decoratedStorage = OtelStorageDecorator.decorate(mockStorage, mockOtel, TransportCompatibility.Transport.HTTP);
+    assertNotNull(decoratedStorage);
+
+    OtelStorageDecorator osd = (OtelStorageDecorator) decoratedStorage;
+    osd.checkCacheAndTriggerFetch("test-poc-bucket");
+
+    BucketMetadataCache.BucketMetadata meta = cache.get("test-poc-bucket");
+    assertNotNull(meta);
+    assertEquals("projects/_/buckets/test-poc-bucket", meta.resource);
+    assertEquals("global", meta.location);
+
+    cache.remove("test-poc-bucket");
+    org.junit.Assert.assertNull(cache.get("test-poc-bucket"));
+  }
+}