@@ -17,9 +17,11 @@
 package com.google.cloud.storage.contrib.nio;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
 
 import com.google.auto.value.AutoValue;
 
+import javax.annotation.Nullable;
 import java.util.Map;
 
 /**
@@ -65,6 +67,25 @@ public abstract class CloudStorageConfiguration {
    */
   public abstract int maxChannelReopens();
 
+  /**
+   * Returns the project to be billed when accessing buckets. Leave empty for normal semantics,
+   * set to bill that project (project you own) for all accesses. This is required for accessing
+   * requester-pays buckets. This value cannot be null.
+   */
+  public abstract @Nullable String userProject();
+
+  /**
+   * Returns whether userProject will be cleared for non-requester-pays buckets. That is,
+   * if false (the default value), setting userProject causes that project to be billed
+   * regardless of whether the bucket is requester-pays or not. If true, setting
+   * userProject will only cause that project to be billed when the project is requester-pays.
+   *
+   * Setting this will cause the bucket to be accessed when the CloudStorageFileSystem object
+   * is created.
+   */
+  public abstract boolean useUserProjectOnlyForRequesterPaysBuckets();
+
+
   /**
    * Creates a new builder, initialized with the following settings:
    *
@@ -90,6 +111,9 @@ public static final class Builder {
     private boolean usePseudoDirectories = true;
     private int blockSize = CloudStorageFileSystem.BLOCK_SIZE_DEFAULT;
     private int maxChannelReopens = 0;
+    private @Nullable String userProject = null;
+    // This of this as "clear userProject if not RequesterPays"
+    private boolean useUserProjectOnlyForRequesterPaysBuckets = false;
 
     /**
      * Changes current working directory for new filesystem. This defaults to the root directory.
@@ -99,6 +123,7 @@ public static final class Builder {
      * @throws IllegalArgumentException if {@code path} is not absolute.
      */
     public Builder workingDirectory(String path) {
+      checkNotNull(path);
       checkArgument(UnixPath.getPath(false, path).isAbsolute(), "not absolute: %s", path);
       workingDirectory = path;
       return this;
@@ -147,6 +172,16 @@ public Builder maxChannelReopens(int value) {
       return this;
     }
 
+    public Builder userProject(String value) {
+      userProject = value;
+      return this;
+    }
+
+    public Builder autoDetectRequesterPays(boolean value) {
+      useUserProjectOnlyForRequesterPaysBuckets = value;
+      return this;
+    }
+
     /**
      * Creates new instance without destroying builder.
      */
@@ -157,7 +192,9 @@ public CloudStorageConfiguration build() {
           stripPrefixSlash,
           usePseudoDirectories,
           blockSize,
-          maxChannelReopens);
+          maxChannelReopens,
+          userProject,
+          useUserProjectOnlyForRequesterPaysBuckets);
     }
 
     Builder(CloudStorageConfiguration toModify) {
@@ -167,6 +204,8 @@ public CloudStorageConfiguration build() {
       usePseudoDirectories = toModify.usePseudoDirectories();
       blockSize = toModify.blockSize();
       maxChannelReopens = toModify.maxChannelReopens();
+      userProject = toModify.userProject();
+      useUserProjectOnlyForRequesterPaysBuckets = toModify.useUserProjectOnlyForRequesterPaysBuckets();
     }
 
     Builder() {}
@@ -201,6 +240,12 @@ static private CloudStorageConfiguration fromMap(Builder builder, Map<String, ?>
         case "maxChannelReopens":
           builder.maxChannelReopens((Integer) entry.getValue());
           break;
+        case "userProject":
+          builder.userProject((String) entry.getValue());
+          break;
+        case "useUserProjectOnlyForRequesterPaysBuckets":
+          builder.autoDetectRequesterPays((Boolean) entry.getValue());
+          break;
         default:
           throw new IllegalArgumentException(entry.getKey());
       }
 
@@ -20,6 +20,7 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import com.google.cloud.storage.StorageOptions;
+import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableSet;
 
 import java.io.IOException;
@@ -33,6 +34,7 @@
 import java.nio.file.WatchService;
 import java.nio.file.attribute.FileTime;
 import java.nio.file.attribute.UserPrincipalLookupService;
+import java.util.HashMap;
 import java.util.Objects;
 import java.util.Set;
 
@@ -112,8 +114,9 @@ public static CloudStorageFileSystem forBucket(String bucket) {
   public static CloudStorageFileSystem forBucket(String bucket, CloudStorageConfiguration config) {
     checkArgument(
         !bucket.startsWith(URI_SCHEME + ":"), "Bucket name must not have schema: %s", bucket);
+    checkNotNull(config);
     return new CloudStorageFileSystem(
-        new CloudStorageFileSystemProvider(), bucket, checkNotNull(config));
+        new CloudStorageFileSystemProvider(config.userProject()), bucket, config);
   }
 
   /**
@@ -136,15 +139,29 @@ public static CloudStorageFileSystem forBucket(String bucket, CloudStorageConfig
       @Nullable StorageOptions storageOptions) {
     checkArgument(!bucket.startsWith(URI_SCHEME + ":"),
         "Bucket name must not have schema: %s", bucket);
-    return new CloudStorageFileSystem(new CloudStorageFileSystemProvider(storageOptions),
+    return new CloudStorageFileSystem(new CloudStorageFileSystemProvider(config.userProject(), storageOptions),
         bucket, checkNotNull(config));
   }
 
   CloudStorageFileSystem(
       CloudStorageFileSystemProvider provider, String bucket, CloudStorageConfiguration config) {
     checkArgument(!bucket.isEmpty(), "bucket");
-    this.provider = provider;
     this.bucket = bucket;
+    if (config.useUserProjectOnlyForRequesterPaysBuckets()) {
+      if (Strings.isNullOrEmpty(config.userProject())) {
+        throw new IllegalArgumentException("If useUserProjectOnlyForRequesterPaysBuckets is set, then userProject must be set too.");
+      }
+      // detect whether we want to pay for these accesses or not.
+      if (!provider.requesterPays(bucket)) {
+        // update config (just to ease debugging, we're not actually using config.userProject later.
+        HashMap<String, String> disableUserProject = new HashMap<>();
+        disableUserProject.put("userProject", "");
+        config = CloudStorageConfiguration.fromMap(config, disableUserProject);
+        // update the provider (this is the most important bit)
+        provider = provider.withNoUserProject();
+      }
+    }
+    this.provider = provider;
     this.config = config;
   }