diff --git a/gcloud-java-pubsub/src/main/java/com/google/cloud/pubsub/PubSub.java b/gcloud-java-pubsub/src/main/java/com/google/cloud/pubsub/PubSub.java
index 428f74b64024..f86817c26345 100644
--- a/gcloud-java-pubsub/src/main/java/com/google/cloud/pubsub/PubSub.java
+++ b/gcloud-java-pubsub/src/main/java/com/google/cloud/pubsub/PubSub.java
@@ -381,8 +381,56 @@ interface MessageConsumer extends AutoCloseable {
    */
   Future<AsyncPage<SubscriptionId>> listSubscriptionsAsync(String topic, ListOption... options);
 
+  /**
+   * Pulls messages from the provided subscription. This method possibly returns no messages if no
+   * message was available at the time the request was processed by the Pub/Sub service (i.e. the
+   * system is not allowed to wait until at least one message is available). Pulled messages have
+   * their acknowledge deadline automatically renewed until they are explicitly consumed using
+   * {@link Iterator#next()}.
+   *
+   * <p>Example usage of synchronous message pulling:
+   * <pre> {@code
+   * Iterator<ReceivedMessage> messageIterator = pubsub.pull("subscription", 100);
+   * while (messageIterator.hasNext()) {
+   *   ReceivedMessage message = messageIterator.next();
+   *   // message's acknowledge deadline is no longer automatically renewed. If processing takes
+   *   // long pubsub.modifyAckDeadline(String, String, long, TimeUnit) can be used to extend it.
+   *   doSomething(message);
+   *   message.ack(); // or message.nack()
+   * }}</pre>
+   *
+   * @param subscription the subscription from which to pull messages
+   * @param maxMessages the maximum number of messages pulled by this method. This method can
+   *     possibly return fewer messages.
+   * @throws PubSubException upon failure
+   */
   Iterator<ReceivedMessage> pull(String subscription, int maxMessages);
 
+  /**
+   * Sends a request for pulling messages from the provided subscription. This method returns a
+   * {@code Future} object to consume the result. {@link Future#get()} returns a message iterator.
+   * This method possibly returns no messages if no message was available at the time the request
+   * was processed by the Pub/Sub service (i.e. the system is not allowed to wait until at least one
+   * message is available).
+   *
+   * <p>Example usage of asynchronous message pulling:
+   * <pre> {@code
+   * Future<Iterator<ReceivedMessage>> future = pubsub.pull("subscription", 100);
+   * // do something while the request gets processed
+   * Iterator<ReceivedMessage> messageIterator = future.get();
+   * while (messageIterator.hasNext()) {
+   *   ReceivedMessage message = messageIterator.next();
+   *   // message's acknowledge deadline is no longer automatically renewed. If processing takes
+   *   // long pubsub.modifyAckDeadline(String, String, long, TimeUnit) can be used to extend it.
+   *   doSomething(message);
+   *   message.ack(); // or message.nack()
+   * }}</pre>
+   *
+   * @param subscription the subscription from which to pull messages
+   * @param maxMessages the maximum number of messages pulled by this method. This method can
+   *     possibly return fewer messages.
+   * @throws PubSubException upon failure
+   */
   Future<Iterator<ReceivedMessage>> pullAsync(String subscription, int maxMessages);
 
   MessageConsumer pullAsync(String subscription, MessageProcessor callback, PullOption... options);
diff --git a/gcloud-java-pubsub/src/main/java/com/google/cloud/pubsub/PubSubImpl.java b/gcloud-java-pubsub/src/main/java/com/google/cloud/pubsub/PubSubImpl.java
index 042f76b5197c..800d26c41ce9 100644
--- a/gcloud-java-pubsub/src/main/java/com/google/cloud/pubsub/PubSubImpl.java
+++ b/gcloud-java-pubsub/src/main/java/com/google/cloud/pubsub/PubSubImpl.java
@@ -16,9 +16,9 @@
 
 package com.google.cloud.pubsub;
 
+import static com.google.api.client.util.Preconditions.checkArgument;
 import static com.google.cloud.pubsub.PubSub.ListOption.OptionType.PAGE_SIZE;
 import static com.google.cloud.pubsub.PubSub.ListOption.OptionType.PAGE_TOKEN;
-import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.util.concurrent.Futures.lazyTransform;
 
 import com.google.cloud.AsyncPage;
@@ -29,10 +29,12 @@
 import com.google.cloud.pubsub.spi.PubSubRpc;
 import com.google.cloud.pubsub.spi.v1.PublisherApi;
 import com.google.cloud.pubsub.spi.v1.SubscriberApi;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Function;
 import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
+import com.google.common.collect.Iterators;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.util.concurrent.Uninterruptibles;
@@ -52,6 +54,8 @@
 import com.google.pubsub.v1.ModifyPushConfigRequest;
 import com.google.pubsub.v1.PublishRequest;
 import com.google.pubsub.v1.PublishResponse;
+import com.google.pubsub.v1.PullRequest;
+import com.google.pubsub.v1.PullResponse;
 
 import java.util.Collections;
 import java.util.Iterator;
@@ -64,6 +68,8 @@
 class PubSubImpl extends BaseService<PubSubOptions> implements PubSub {
 
   private final PubSubRpc rpc;
+  private final AckDeadlineRenewer ackDeadlineRenewer;
+  private boolean closed;
 
   private static final Function<Empty, Void> EMPTY_TO_VOID_FUNCTION = new Function<Empty, Void>() {
     @Override
@@ -78,10 +84,25 @@ public Boolean apply(Empty input) {
           return input != null;
         }
       };
+  private static final Function<com.google.pubsub.v1.ReceivedMessage, String>
+      MESSAGE_TO_ACK_ID_FUNCTION = new Function<com.google.pubsub.v1.ReceivedMessage, String>() {
+        @Override
+        public String apply(com.google.pubsub.v1.ReceivedMessage message) {
+          return message.getAckId();
+        }
+      };
 
   PubSubImpl(PubSubOptions options) {
     super(options);
     rpc = options.rpc();
+    ackDeadlineRenewer = new AckDeadlineRenewer(this);
+  }
+
+  @VisibleForTesting
+  PubSubImpl(PubSubOptions options, AckDeadlineRenewer ackDeadlineRenewer) {
+    super(options);
+    rpc = options.rpc();
+    this.ackDeadlineRenewer = ackDeadlineRenewer;
   }
 
   private abstract static class BasePageFetcher<T> implements AsyncPageImpl.NextPageFetcher<T> {
@@ -445,17 +466,35 @@ public Future<AsyncPage<SubscriptionId>> listSubscriptionsAsync(String topic,
 
   @Override
   public Iterator<ReceivedMessage> pull(String subscription, int maxMessages) {
-    // this should set return_immediately to true
-    return null;
+    return get(pullAsync(subscription, maxMessages));
   }
 
   @Override
-  public Future<Iterator<ReceivedMessage>> pullAsync(String subscription, int maxMessages) {
-    // though this method can set return_immediately to false (as future can be canceled) I
-    // suggest to keep it false so sync could delegate to asyc and use the same options
-    // this method also should use the VTKIT thread-pool to renew ack deadline for non consumed
-    // messages
-    return null;
+  public Future<Iterator<ReceivedMessage>> pullAsync(final String subscription, int maxMessages) {
+    PullRequest request = PullRequest.newBuilder().setReturnImmediately(true)
+        .setSubscription(SubscriberApi.formatSubscriptionName(options().projectId(), subscription))
+        .setMaxMessages(maxMessages)
+        .setReturnImmediately(true)
+        .build();
+    Future<PullResponse> response = rpc.pull(request);
+    return lazyTransform(response, new Function<PullResponse, Iterator<ReceivedMessage>>() {
+      @Override
+      public Iterator<ReceivedMessage> apply(PullResponse pullResponse) {
+        // Add all received messages to the automatic ack deadline renewer
+        List<String> ackIds = Lists.transform(pullResponse.getReceivedMessagesList(),
+            MESSAGE_TO_ACK_ID_FUNCTION);
+        ackDeadlineRenewer.add(subscription, ackIds);
+        return Iterators.transform(pullResponse.getReceivedMessagesList().iterator(),
+            new Function<com.google.pubsub.v1.ReceivedMessage, ReceivedMessage>() {
+              @Override
+              public ReceivedMessage apply(com.google.pubsub.v1.ReceivedMessage receivedMessage) {
+                // Remove consumed message from automatic ack deadline renewer
+                ackDeadlineRenewer.remove(subscription, receivedMessage.getAckId());
+                return ReceivedMessage.fromPb(PubSubImpl.this, subscription, receivedMessage);
+              }
+            });
+      }
+    });
   }
 
   @Override
@@ -549,6 +588,13 @@ public Future<Void> modifyAckDeadlineAsync(String subscription, int deadline, Ti
 
   @Override
   public void close() throws Exception {
+    if (closed) {
+      return;
+    }
+    closed = true;
     rpc.close();
+    if (ackDeadlineRenewer != null) {
+      ackDeadlineRenewer.close();
+    }
   }
 }
diff --git a/gcloud-java-pubsub/src/main/java/com/google/cloud/pubsub/spi/DefaultPubSubRpc.java b/gcloud-java-pubsub/src/main/java/com/google/cloud/pubsub/spi/DefaultPubSubRpc.java
index 7df6cb2632f5..4ac3837088f8 100644
--- a/gcloud-java-pubsub/src/main/java/com/google/cloud/pubsub/spi/DefaultPubSubRpc.java
+++ b/gcloud-java-pubsub/src/main/java/com/google/cloud/pubsub/spi/DefaultPubSubRpc.java
@@ -73,6 +73,8 @@ public class DefaultPubSubRpc implements PubSubRpc {
   private final ScheduledExecutorService executor;
   private final ExecutorFactory executorFactory;
 
+  private boolean closed;
+
   private static final class InternalPubSubOptions extends PubSubOptions {
 
     private static final long serialVersionUID = -7997372049256706185L;
@@ -233,6 +235,10 @@ public Future<Empty> modify(ModifyPushConfigRequest request) {
 
   @Override
   public void close() throws Exception {
+    if (closed) {
+      return;
+    }
+    closed = true;
     subscriberApi.close();
     publisherApi.close();
     executorFactory.release(executor);
diff --git a/gcloud-java-pubsub/src/test/java/com/google/cloud/pubsub/AckDeadlineRenewerTest.java b/gcloud-java-pubsub/src/test/java/com/google/cloud/pubsub/AckDeadlineRenewerTest.java
index 07cc70c6e30d..bf912e0a79f7 100644
--- a/gcloud-java-pubsub/src/test/java/com/google/cloud/pubsub/AckDeadlineRenewerTest.java
+++ b/gcloud-java-pubsub/src/test/java/com/google/cloud/pubsub/AckDeadlineRenewerTest.java
@@ -25,7 +25,9 @@
 import org.easymock.IAnswer;
 import org.junit.After;
 import org.junit.Before;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.Timeout;
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
@@ -47,6 +49,9 @@ public class AckDeadlineRenewerTest {
   private PubSub pubsub;
   private AckDeadlineRenewer ackDeadlineRenewer;
 
+  @Rule
+  public Timeout globalTimeout = Timeout.seconds(60);
+
   @Before
   public void setUp() {
     pubsub = EasyMock.createStrictMock(PubSub.class);
diff --git a/gcloud-java-pubsub/src/test/java/com/google/cloud/pubsub/BaseSystemTest.java b/gcloud-java-pubsub/src/test/java/com/google/cloud/pubsub/BaseSystemTest.java
index 3360632b45ce..8eb03a7de073 100644
--- a/gcloud-java-pubsub/src/test/java/com/google/cloud/pubsub/BaseSystemTest.java
+++ b/gcloud-java-pubsub/src/test/java/com/google/cloud/pubsub/BaseSystemTest.java
@@ -22,6 +22,7 @@
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
+import com.google.api.client.util.Lists;
 import com.google.cloud.AsyncPage;
 import com.google.cloud.Page;
 import com.google.common.collect.ImmutableList;
@@ -37,6 +38,7 @@
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
 /**
  * A base class for system tests. This class can be extended to run system tests in different
@@ -428,4 +430,218 @@ public void testListSubscriptionsAsync() throws ExecutionException, InterruptedE
     assertTrue(subscription2.delete());
     assertTrue(subscription3.delete());
   }
+
+  @Test
+  public void testPullMessages() {
+    String topic = formatForTest("test-pull-messages-topic");
+    pubsub().create(TopicInfo.of(topic));
+    String subscription = formatForTest("test-pull-messages-subscription");
+    pubsub().create(SubscriptionInfo.of(topic, subscription));
+    Message message1 = Message.of("payload1");
+    Message message2 = Message.of("payload2");
+    List<String> messageIds = pubsub().publish(topic, ImmutableList.of(message1, message2));
+    assertEquals(2, messageIds.size());
+    Iterator<ReceivedMessage> iterator = pubsub().pull(subscription, 2);
+    assertEquals(message1.payloadAsString(), iterator.next().payloadAsString());
+    assertEquals(message2.payloadAsString(), iterator.next().payloadAsString());
+    assertTrue(pubsub().deleteSubscription(subscription));
+    assertTrue(pubsub().deleteTopic(topic));
+  }
+
+  @Test
+  public void testPullMessagesAndAutoRenewDeadline() throws InterruptedException {
+    String topic = formatForTest("test-pull-messages-and-renew-deadline-topic");
+    pubsub().create(TopicInfo.of(topic));
+    String subscription = formatForTest("test-pull-messages-and-renew-deadline-subscription");
+    pubsub().create(SubscriptionInfo.builder(topic, subscription).ackDeadLineSeconds(10).build());
+    Message message1 = Message.of("payload1");
+    Message message2 = Message.of("payload2");
+    // todo(mziccard): use batch publish if #1017 gets fixed, or remove this comment
+    pubsub().publish(topic, message1);
+    pubsub().publish(topic, message2);
+    Iterator<ReceivedMessage> iterator = pubsub().pull(subscription, 2);
+    ReceivedMessage consumedMessage = iterator.next();
+    Thread.sleep(15000);
+    // first message was consumed while second message is still being renewed
+    Iterator<ReceivedMessage> nextIterator = pubsub().pull(subscription, 2);
+    assertTrue(nextIterator.hasNext());
+    ReceivedMessage message = nextIterator.next();
+    assertEquals(consumedMessage.payloadAsString(), message.payloadAsString());
+    assertFalse(nextIterator.hasNext());
+    consumedMessage.ack();
+    iterator.next().ack();
+    nextIterator = pubsub().pull(subscription, 2);
+    assertFalse(nextIterator.hasNext());
+    assertTrue(pubsub().deleteSubscription(subscription));
+    assertTrue(pubsub().deleteTopic(topic));
+  }
+
+  @Test
+  public void testPullMessagesAndModifyAckDeadline() throws InterruptedException {
+    String topic = formatForTest("test-pull-messages-and-modify-deadline-topic");
+    pubsub().create(TopicInfo.of(topic));
+    String subscription = formatForTest("test-pull-messages-and-modify-deadline-subscription");
+    pubsub().create(SubscriptionInfo.builder(topic, subscription).ackDeadLineSeconds(10).build());
+    Message message1 = Message.of("payload1");
+    Message message2 = Message.of("payload2");
+    // todo(mziccard): use batch publish if #1017 gets fixed, or remove this comment
+    pubsub().publish(topic, message1);
+    pubsub().publish(topic, message2);
+    // Consume all messages and stop ack renewal
+    List<ReceivedMessage> receivedMessages = Lists.newArrayList(pubsub().pull(subscription, 2));
+    receivedMessages.get(0).modifyAckDeadline(60, TimeUnit.SECONDS);
+    Thread.sleep(15000);
+    // first message was renewed while second message should still be sent on pull requests
+    Iterator<ReceivedMessage> nextIterator = pubsub().pull(subscription, 2);
+    assertTrue(nextIterator.hasNext());
+    ReceivedMessage message = nextIterator.next();
+    assertEquals(receivedMessages.get(1).payloadAsString(), message.payloadAsString());
+    assertFalse(nextIterator.hasNext());
+    assertTrue(pubsub().deleteSubscription(subscription));
+    assertTrue(pubsub().deleteTopic(topic));
+  }
+
+  @Test
+  public void testPullNonExistingSubscription() {
+    thrown.expect(PubSubException.class);
+    pubsub().pull(formatForTest("non-existing-subscription"), 2);
+  }
+
+  @Test
+  public void testPullMessagesAsync() throws ExecutionException, InterruptedException {
+    String topic = formatForTest("test-pull-messages-async-topic");
+    pubsub().create(TopicInfo.of(topic));
+    String subscription = formatForTest("test-pull-messages-async-subscription");
+    pubsub().create(SubscriptionInfo.of(topic, subscription));
+    Message message1 = Message.of("payload1");
+    Message message2 = Message.of("payload2");
+    List<String> messageIds = pubsub().publish(topic, ImmutableList.of(message1, message2));
+    assertEquals(2, messageIds.size());
+    Iterator<ReceivedMessage> iterator = pubsub().pullAsync(subscription, 2).get();
+    assertEquals(message1.payloadAsString(), iterator.next().payloadAsString());
+    assertEquals(message2.payloadAsString(), iterator.next().payloadAsString());
+    assertTrue(pubsub().deleteSubscription(subscription));
+    assertTrue(pubsub().deleteTopic(topic));
+  }
+
+  @Test
+  public void testPullAsyncNonExistingSubscription()
+      throws ExecutionException, InterruptedException {
+    thrown.expect(ExecutionException.class);
+    pubsub().pullAsync(formatForTest("non-existing-subscription"), 2).get();
+  }
+
+  @Test
+  public void testAckAndNackOneMessage() {
+    String topic = formatForTest("test-ack-one-message-topic");
+    pubsub().create(TopicInfo.of(topic));
+    String subscription = formatForTest("test-ack-one-message-subscription");
+    pubsub().create(SubscriptionInfo.of(topic, subscription));
+    Message message = Message.of("payload");
+    assertNotNull(pubsub().publish(topic, message));
+    Iterator<ReceivedMessage> receivedMessages = pubsub().pull(subscription, 1);
+    receivedMessages.next().nack();
+    receivedMessages = pubsub().pull(subscription, 1);
+    receivedMessages.next().ack();
+    assertFalse(pubsub().pull(subscription, 1).hasNext());
+    assertTrue(pubsub().deleteSubscription(subscription));
+    assertTrue(pubsub().deleteTopic(topic));
+  }
+
+  @Test
+  public void testAckAndNackOneMessageAsync() throws ExecutionException, InterruptedException {
+    String topic = formatForTest("test-ack-one-message-async-topic");
+    pubsub().create(TopicInfo.of(topic));
+    String subscription = formatForTest("test-ack-one-message-async-subscription");
+    pubsub().create(SubscriptionInfo.of(topic, subscription));
+    Message message = Message.of("payload");
+    assertNotNull(pubsub().publish(topic, message));
+    Iterator<ReceivedMessage> receivedMessages = pubsub().pull(subscription, 1);
+    receivedMessages.next().nackAsync().get();
+    receivedMessages = pubsub().pull(subscription, 1);
+    receivedMessages.next().ackAsync().get();
+    assertFalse(pubsub().pull(subscription, 1).hasNext());
+    assertTrue(pubsub().deleteSubscription(subscription));
+    assertTrue(pubsub().deleteTopic(topic));
+  }
+
+  @Test
+  public void testAckAndNackMoreMessages() throws ExecutionException, InterruptedException {
+    String topic = formatForTest("test-ack-more-messages-topic");
+    pubsub().create(TopicInfo.of(topic));
+    String subscription = formatForTest("test-ack-more-messages-subscription");
+    pubsub().create(SubscriptionInfo.of(topic, subscription));
+    Message message1 = Message.of("payload1");
+    Message message2 = Message.of("payload2");
+    assertNotNull(pubsub().publish(topic, message1, message2));
+    Iterator<ReceivedMessage> receivedMessages = pubsub().pull(subscription, 2);
+    pubsub().nack(subscription, receivedMessages.next().ackId(), receivedMessages.next().ackId());
+    receivedMessages = pubsub().pull(subscription, 2);
+    pubsub().ack(subscription, receivedMessages.next().ackId(), receivedMessages.next().ackId());
+    assertFalse(pubsub().pull(subscription, 2).hasNext());
+    assertTrue(pubsub().deleteSubscription(subscription));
+    assertTrue(pubsub().deleteTopic(topic));
+  }
+
+  @Test
+  public void testAckAndNackMoreMessagesAsync() throws ExecutionException, InterruptedException {
+    String topic = formatForTest("test-ack-more-messages-async-topic");
+    pubsub().create(TopicInfo.of(topic));
+    String subscription = formatForTest("test-ack-more-messages-async-subscription");
+    pubsub().create(SubscriptionInfo.of(topic, subscription));
+    Message message1 = Message.of("payload1");
+    Message message2 = Message.of("payload2");
+    assertNotNull(pubsub().publish(topic, message1, message2));
+    Iterator<ReceivedMessage> receivedMessages = pubsub().pull(subscription, 2);
+    pubsub()
+        .nackAsync(subscription, receivedMessages.next().ackId(), receivedMessages.next().ackId())
+        .get();
+    receivedMessages = pubsub().pull(subscription, 2);
+    pubsub()
+        .ackAsync(subscription, receivedMessages.next().ackId(), receivedMessages.next().ackId())
+        .get();
+    assertFalse(pubsub().pull(subscription, 2).hasNext());
+    assertTrue(pubsub().deleteSubscription(subscription));
+    assertTrue(pubsub().deleteTopic(topic));
+  }
+
+  @Test
+  public void testAckAndNackMessageList() throws ExecutionException, InterruptedException {
+    String topic = formatForTest("test-ack-message-list-topic");
+    pubsub().create(TopicInfo.of(topic));
+    String subscription = formatForTest("test-ack-message-list-subscription");
+    pubsub().create(SubscriptionInfo.of(topic, subscription));
+    Message message1 = Message.of("payload1");
+    Message message2 = Message.of("payload2");
+    assertNotNull(pubsub().publish(topic, ImmutableList.of(message1, message2)));
+    Iterator<ReceivedMessage> receivedMessages = pubsub().pull(subscription, 2);
+    pubsub().nack(subscription,
+        ImmutableList.of(receivedMessages.next().ackId(), receivedMessages.next().ackId()));
+    receivedMessages = pubsub().pull(subscription, 2);
+    pubsub().ack(subscription,
+        ImmutableList.of(receivedMessages.next().ackId(), receivedMessages.next().ackId()));
+    assertFalse(pubsub().pull(subscription, 2).hasNext());
+    assertTrue(pubsub().deleteSubscription(subscription));
+    assertTrue(pubsub().deleteTopic(topic));
+  }
+
+  @Test
+  public void testAckAndNackMessageListAsync() throws ExecutionException, InterruptedException {
+    String topic = formatForTest("test-ack-message-list-async-topic");
+    pubsub().create(TopicInfo.of(topic));
+    String subscription = formatForTest("test-ack-message-list-async-subscription");
+    pubsub().create(SubscriptionInfo.of(topic, subscription));
+    Message message1 = Message.of("payload1");
+    Message message2 = Message.of("payload2");
+    assertNotNull(pubsub().publish(topic, ImmutableList.of(message1, message2)));
+    Iterator<ReceivedMessage> receivedMessages = pubsub().pull(subscription, 2);
+    pubsub().nackAsync(subscription,
+        ImmutableList.of(receivedMessages.next().ackId(), receivedMessages.next().ackId())).get();
+    receivedMessages = pubsub().pull(subscription, 2);
+    pubsub().ackAsync(subscription,
+        ImmutableList.of(receivedMessages.next().ackId(), receivedMessages.next().ackId())).get();
+    assertFalse(pubsub().pull(subscription, 2).hasNext());
+    assertTrue(pubsub().deleteSubscription(subscription));
+    assertTrue(pubsub().deleteTopic(topic));
+  }
 }
diff --git a/gcloud-java-pubsub/src/test/java/com/google/cloud/pubsub/PubSubImplTest.java b/gcloud-java-pubsub/src/test/java/com/google/cloud/pubsub/PubSubImplTest.java
index 6084266492f0..7a9fee0d999d 100644
--- a/gcloud-java-pubsub/src/test/java/com/google/cloud/pubsub/PubSubImplTest.java
+++ b/gcloud-java-pubsub/src/test/java/com/google/cloud/pubsub/PubSubImplTest.java
@@ -52,6 +52,8 @@
 import com.google.pubsub.v1.ModifyPushConfigRequest;
 import com.google.pubsub.v1.PublishRequest;
 import com.google.pubsub.v1.PublishResponse;
+import com.google.pubsub.v1.PullRequest;
+import com.google.pubsub.v1.PullResponse;
 
 import org.easymock.EasyMock;
 import org.junit.After;
@@ -60,6 +62,7 @@
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
 
+import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
@@ -93,6 +96,18 @@ public com.google.pubsub.v1.Topic apply(TopicInfo topicInfo) {
           .ackDeadLineSeconds(42)
           .pushConfig(PUSH_CONFIG)
           .build();
+  private static final Message MESSAGE1 = Message.of("payload1");
+  private static final com.google.pubsub.v1.ReceivedMessage MESSAGE_PB1 =
+      com.google.pubsub.v1.ReceivedMessage.newBuilder()
+          .setMessage(MESSAGE1.toPb())
+          .setAckId("ackId1")
+          .build();
+  private static final Message MESSAGE2 = Message.of("payload2");
+  private static final com.google.pubsub.v1.ReceivedMessage MESSAGE_PB2 =
+      com.google.pubsub.v1.ReceivedMessage.newBuilder()
+          .setMessage(MESSAGE2.toPb())
+          .setAckId("ackId2")
+          .build();
   private static final Function<SubscriptionInfo, com.google.pubsub.v1.Subscription>
       SUBSCRIPTION_TO_PB_FUNCTION =
           new Function<SubscriptionInfo, com.google.pubsub.v1.Subscription>() {
@@ -112,6 +127,7 @@ public String apply(SubscriptionId subscriptionId) {
   private PubSubOptions options;
   private PubSubRpcFactory rpcFactoryMock;
   private PubSubRpc pubsubRpcMock;
+  private AckDeadlineRenewer renewerMock;
   private PubSub pubsub;
 
   @Rule
@@ -121,26 +137,32 @@ public String apply(SubscriptionId subscriptionId) {
   public void setUp() {
     rpcFactoryMock = EasyMock.createStrictMock(PubSubRpcFactory.class);
     pubsubRpcMock = EasyMock.createStrictMock(PubSubRpc.class);
-    EasyMock.expect(rpcFactoryMock.create(EasyMock.anyObject(PubSubOptions.class)))
-        .andReturn(pubsubRpcMock).times(1);
-    options = PubSubOptions.builder()
-        .projectId(PROJECT)
-        .serviceRpcFactory(rpcFactoryMock)
-        .retryParams(RetryParams.noRetries())
-        .build();
-    EasyMock.replay(rpcFactoryMock, pubsubRpcMock);
-    EasyMock.reset(pubsubRpcMock);
+    renewerMock = EasyMock.createStrictMock(AckDeadlineRenewer.class);
+    options = EasyMock.createMock(PubSubOptions.class);
+    EasyMock.expect(options.projectId()).andReturn(PROJECT).anyTimes();
+    EasyMock.expect(options.rpc()).andReturn(pubsubRpcMock).anyTimes();
+    EasyMock.expect(options.retryParams()).andReturn(RetryParams.noRetries()).anyTimes();
+    EasyMock.replay(rpcFactoryMock, pubsubRpcMock, renewerMock, options);
+    EasyMock.reset(pubsubRpcMock, renewerMock);
   }
 
   @After
   public void tearDown() {
-    EasyMock.verify(rpcFactoryMock, pubsubRpcMock);
+    EasyMock.verify(rpcFactoryMock, pubsubRpcMock, renewerMock, options);
+  }
+
+  private void resetOptionsForList(int pageCount) {
+    EasyMock.reset(options);
+    EasyMock.expect(options.projectId()).andReturn(PROJECT).times(pageCount);
+    EasyMock.expect(options.rpc()).andReturn(pubsubRpcMock).times(pageCount);
+    EasyMock.expect(options.service()).andReturn(pubsub).times(pageCount);
+    EasyMock.replay(options);
   }
 
   @Test
   public void testGetOptions() {
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertSame(options, pubsub.options());
   }
 
@@ -149,8 +171,8 @@ public void testCreateTopic() {
     com.google.pubsub.v1.Topic topicPb = TOPIC_INFO.toPb(PROJECT);
     Future<com.google.pubsub.v1.Topic> response = Futures.immediateFuture(topicPb);
     EasyMock.expect(pubsubRpcMock.create(topicPb)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     Topic topic = pubsub.create(TOPIC_INFO);
     assertEquals(new Topic(pubsub, new TopicInfo.BuilderImpl(TOPIC_INFO)), topic);
   }
@@ -160,8 +182,8 @@ public void testCreateTopicAsync() throws ExecutionException, InterruptedExcepti
     com.google.pubsub.v1.Topic topicPb = TOPIC_INFO.toPb(PROJECT);
     Future<com.google.pubsub.v1.Topic> response = Futures.immediateFuture(topicPb);
     EasyMock.expect(pubsubRpcMock.create(topicPb)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     Topic topic = pubsub.createAsync(TOPIC_INFO).get();
     assertEquals(new Topic(pubsub, new TopicInfo.BuilderImpl(TOPIC_INFO)), topic);
   }
@@ -172,8 +194,8 @@ public void testGetTopic() {
     Future<com.google.pubsub.v1.Topic> response =
         Futures.immediateFuture(TOPIC_INFO.toPb(PROJECT));
     EasyMock.expect(pubsubRpcMock.get(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     Topic topic = pubsub.getTopic(TOPIC);
     assertEquals(new Topic(pubsub, new TopicInfo.BuilderImpl(TOPIC_INFO)), topic);
   }
@@ -183,8 +205,8 @@ public void testGetTopic_Null() {
     GetTopicRequest request = GetTopicRequest.newBuilder().setTopic(TOPIC_NAME_PB).build();
     Future<com.google.pubsub.v1.Topic> responseFuture = Futures.immediateFuture(null);
     EasyMock.expect(pubsubRpcMock.get(request)).andReturn(responseFuture);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertNull(pubsub.getTopic(TOPIC));
   }
 
@@ -194,8 +216,8 @@ public void testGetTopicAsync() throws ExecutionException, InterruptedException
     Future<com.google.pubsub.v1.Topic> response =
         Futures.immediateFuture(TOPIC_INFO.toPb(PROJECT));
     EasyMock.expect(pubsubRpcMock.get(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     Future<Topic> topicFuture = pubsub.getTopicAsync(TOPIC);
     assertEquals(new Topic(pubsub, new TopicInfo.BuilderImpl(TOPIC_INFO)), topicFuture.get());
   }
@@ -205,8 +227,8 @@ public void testGetTopicAsync_Null() throws ExecutionException, InterruptedExcep
     GetTopicRequest request = GetTopicRequest.newBuilder().setTopic(TOPIC_NAME_PB).build();
     Future<com.google.pubsub.v1.Topic> responseFuture = Futures.immediateFuture(null);
     EasyMock.expect(pubsubRpcMock.get(request)).andReturn(responseFuture);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertNull(pubsub.getTopicAsync(TOPIC).get());
   }
 
@@ -215,8 +237,8 @@ public void testDeleteTopic() {
     DeleteTopicRequest request = DeleteTopicRequest.newBuilder().setTopic(TOPIC_NAME_PB).build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.delete(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertTrue(pubsub.deleteTopic(TOPIC));
   }
 
@@ -225,8 +247,8 @@ public void testDeleteTopic_Null() {
     DeleteTopicRequest request = DeleteTopicRequest.newBuilder().setTopic(TOPIC_NAME_PB).build();
     Future<Empty> response = Futures.immediateFuture(null);
     EasyMock.expect(pubsubRpcMock.delete(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertFalse(pubsub.deleteTopic(TOPIC));
   }
 
@@ -235,8 +257,8 @@ public void testDeleteTopicAsync() throws ExecutionException, InterruptedExcepti
     DeleteTopicRequest request = DeleteTopicRequest.newBuilder().setTopic(TOPIC_NAME_PB).build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.delete(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertTrue(pubsub.deleteTopicAsync(TOPIC).get());
   }
 
@@ -245,15 +267,16 @@ public void testDeleteTopicAsync_Null() throws ExecutionException, InterruptedEx
     DeleteTopicRequest request = DeleteTopicRequest.newBuilder().setTopic(TOPIC_NAME_PB).build();
     Future<Empty> response = Futures.immediateFuture(null);
     EasyMock.expect(pubsubRpcMock.delete(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertFalse(pubsub.deleteTopicAsync(TOPIC).get());
   }
 
   @Test
   public void testListTopics() {
     String cursor = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(1);
     ListTopicsRequest request = ListTopicsRequest.newBuilder().setProject(PROJECT_PB).build();
     List<Topic> topicList = ImmutableList.of(
         new Topic(pubsub, new TopicInfo.BuilderImpl(TOPIC_INFO)),
@@ -264,7 +287,7 @@ public void testListTopics() {
         .build();
     Future<ListTopicsResponse> futureResponse = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Page<Topic> page = pubsub.listTopics();
     assertEquals(cursor, page.nextPageCursor());
     assertArrayEquals(topicList.toArray(), Iterables.toArray(page.values(), Topic.class));
@@ -273,7 +296,8 @@ public void testListTopics() {
   @Test
   public void testListTopicsNextPage() {
     String cursor1 = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(2);
     ListTopicsRequest request1 = ListTopicsRequest.newBuilder().setProject(PROJECT_PB).build();
     ListTopicsRequest request2 = ListTopicsRequest.newBuilder()
         .setProject(PROJECT_PB)
@@ -297,7 +321,7 @@ public void testListTopicsNextPage() {
     Future<ListTopicsResponse> futureResponse2 = Futures.immediateFuture(response2);
     EasyMock.expect(pubsubRpcMock.list(request1)).andReturn(futureResponse1);
     EasyMock.expect(pubsubRpcMock.list(request2)).andReturn(futureResponse2);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Page<Topic> page = pubsub.listTopics();
     assertEquals(cursor1, page.nextPageCursor());
     assertArrayEquals(topicList1.toArray(), Iterables.toArray(page.values(), Topic.class));
@@ -308,7 +332,8 @@ public void testListTopicsNextPage() {
 
   @Test
   public void testListTopicsEmpty() {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(1);
     ListTopicsRequest request = ListTopicsRequest.newBuilder().setProject(PROJECT_PB).build();
     List<Topic> topicList = ImmutableList.of();
     ListTopicsResponse response = ListTopicsResponse.newBuilder()
@@ -317,7 +342,7 @@ public void testListTopicsEmpty() {
         .build();
     Future<ListTopicsResponse> futureResponse = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Page<Topic> page = pubsub.listTopics();
     assertNull(page.nextPageCursor());
     assertNull(page.nextPage());
@@ -327,7 +352,8 @@ public void testListTopicsEmpty() {
   @Test
   public void testListTopicsWithOptions() {
     String cursor = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(1);
     ListTopicsRequest request = ListTopicsRequest.newBuilder()
         .setProject(PROJECT_PB)
         .setPageSize(42)
@@ -342,7 +368,7 @@ public void testListTopicsWithOptions() {
         .build();
     Future<ListTopicsResponse> futureResponse = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Page<Topic> page = pubsub.listTopics(ListOption.pageSize(42), ListOption.pageToken(cursor));
     assertNull(page.nextPageCursor());
     assertNull(page.nextPage());
@@ -352,7 +378,8 @@ public void testListTopicsWithOptions() {
   @Test
   public void testListTopicsAsync() throws ExecutionException, InterruptedException {
     String cursor = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(1);
     ListTopicsRequest request = ListTopicsRequest.newBuilder().setProject(PROJECT_PB).build();
     List<Topic> topicList = ImmutableList.of(
         new Topic(pubsub, new TopicInfo.BuilderImpl(TOPIC_INFO)),
@@ -363,7 +390,7 @@ public void testListTopicsAsync() throws ExecutionException, InterruptedExceptio
         .build();
     Future<ListTopicsResponse> futureResponse = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     AsyncPage<Topic> page = pubsub.listTopicsAsync().get();
     assertEquals(cursor, page.nextPageCursor());
     assertArrayEquals(topicList.toArray(), Iterables.toArray(page.values(), Topic.class));
@@ -372,7 +399,8 @@ public void testListTopicsAsync() throws ExecutionException, InterruptedExceptio
   @Test
   public void testListTopicsAsyncNextPage() throws ExecutionException, InterruptedException {
     String cursor1 = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(2);
     ListTopicsRequest request1 = ListTopicsRequest.newBuilder().setProject(PROJECT_PB).build();
     ListTopicsRequest request2 = ListTopicsRequest.newBuilder()
         .setProject(PROJECT_PB)
@@ -396,7 +424,7 @@ public void testListTopicsAsyncNextPage() throws ExecutionException, Interrupted
     Future<ListTopicsResponse> futureResponse2 = Futures.immediateFuture(response2);
     EasyMock.expect(pubsubRpcMock.list(request1)).andReturn(futureResponse1);
     EasyMock.expect(pubsubRpcMock.list(request2)).andReturn(futureResponse2);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     AsyncPage<Topic> page = pubsub.listTopicsAsync().get();
     assertEquals(cursor1, page.nextPageCursor());
     assertArrayEquals(topicList1.toArray(), Iterables.toArray(page.values(), Topic.class));
@@ -407,7 +435,8 @@ public void testListTopicsAsyncNextPage() throws ExecutionException, Interrupted
 
   @Test
   public void testListTopicsAsyncEmpty() throws ExecutionException, InterruptedException {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(1);
     ListTopicsRequest request = ListTopicsRequest.newBuilder().setProject(PROJECT_PB).build();
     List<Topic> topicList = ImmutableList.of();
     ListTopicsResponse response = ListTopicsResponse.newBuilder()
@@ -416,7 +445,7 @@ public void testListTopicsAsyncEmpty() throws ExecutionException, InterruptedExc
         .build();
     Future<ListTopicsResponse> futureResponse = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     AsyncPage<Topic> page = pubsub.listTopicsAsync().get();
     assertNull(page.nextPageCursor());
     assertNull(page.nextPageAsync().get());
@@ -427,7 +456,8 @@ public void testListTopicsAsyncEmpty() throws ExecutionException, InterruptedExc
   @Test
   public void testListTopicsAsyncWithOptions() throws ExecutionException, InterruptedException {
     String cursor = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(1);
     ListTopicsRequest request = ListTopicsRequest.newBuilder()
         .setProject(PROJECT_PB)
         .setPageSize(42)
@@ -442,7 +472,7 @@ public void testListTopicsAsyncWithOptions() throws ExecutionException, Interrup
         .build();
     Future<ListTopicsResponse> futureResponse = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     AsyncPage<Topic> page =
         pubsub.listTopicsAsync(ListOption.pageSize(42), ListOption.pageToken(cursor)).get();
     assertNull(page.nextPageCursor());
@@ -460,8 +490,8 @@ public void testPublishOneMessage() {
     PublishResponse response = PublishResponse.newBuilder().addMessageIds(messageId).build();
     Future<PublishResponse> responseFuture = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.publish(request)).andReturn(responseFuture);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertEquals(messageId, pubsub.publish(TOPIC, MESSAGE));
   }
 
@@ -475,8 +505,8 @@ public void testPublishOneMessageAsync() throws ExecutionException, InterruptedE
     PublishResponse response = PublishResponse.newBuilder().addMessageIds(messageId).build();
     Future<PublishResponse> responseFuture = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.publish(request)).andReturn(responseFuture);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertEquals(messageId, pubsub.publishAsync(TOPIC, MESSAGE).get());
   }
 
@@ -492,8 +522,8 @@ public void testPublishMoreMessages() {
         .build();
     Future<PublishResponse> responseFuture = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.publish(request)).andReturn(responseFuture);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertEquals(messageIds, pubsub.publish(TOPIC, MESSAGE, MESSAGE));
   }
 
@@ -509,8 +539,8 @@ public void testPublishMoreMessagesAsync() throws ExecutionException, Interrupte
         .build();
     Future<PublishResponse> responseFuture = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.publish(request)).andReturn(responseFuture);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertEquals(messageIds, pubsub.publishAsync(TOPIC, MESSAGE, MESSAGE).get());
   }
 
@@ -526,8 +556,8 @@ public void testPublishMessageList() {
         .build();
     Future<PublishResponse> responseFuture = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.publish(request)).andReturn(responseFuture);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertEquals(messageIds, pubsub.publish(TOPIC, ImmutableList.of(MESSAGE, MESSAGE)));
   }
 
@@ -543,8 +573,8 @@ public void testPublishMessageListAsync() throws ExecutionException, Interrupted
         .build();
     Future<PublishResponse> responseFuture = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.publish(request)).andReturn(responseFuture);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertEquals(messageIds, pubsub.publishAsync(TOPIC, ImmutableList.of(MESSAGE, MESSAGE)).get());
   }
 
@@ -554,8 +584,8 @@ public void testCreateSubscription() {
     Future<com.google.pubsub.v1.Subscription> response =
         Futures.immediateFuture(subscriptionPb);
     EasyMock.expect(pubsubRpcMock.create(subscriptionPb)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     Subscription subscription = pubsub.create(SUBSCRIPTION_INFO);
     assertEquals(
         new Subscription(pubsub, new SubscriptionInfo.BuilderImpl(COMPLETE_SUBSCRIPTION_INFO)),
@@ -568,8 +598,8 @@ public void testCreateSubscriptionAsync() throws ExecutionException, Interrupted
     Future<com.google.pubsub.v1.Subscription> response =
         Futures.immediateFuture(subscriptionPb);
     EasyMock.expect(pubsubRpcMock.create(subscriptionPb)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     Subscription subscription = pubsub.createAsync(SUBSCRIPTION_INFO).get();
     assertEquals(
         new Subscription(pubsub, new SubscriptionInfo.BuilderImpl(COMPLETE_SUBSCRIPTION_INFO)),
@@ -583,8 +613,8 @@ public void testGetSubscription() {
     Future<com.google.pubsub.v1.Subscription> response =
         Futures.immediateFuture(SUBSCRIPTION_INFO.toPb(PROJECT));
     EasyMock.expect(pubsubRpcMock.get(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     Subscription subscription = pubsub.getSubscription(SUBSCRIPTION);
     assertEquals(
         new Subscription(pubsub, new SubscriptionInfo.BuilderImpl(COMPLETE_SUBSCRIPTION_INFO)),
@@ -597,8 +627,8 @@ public void testGetSubscription_Null() {
         GetSubscriptionRequest.newBuilder().setSubscription(SUBSCRIPTION_NAME_PB).build();
     Future<com.google.pubsub.v1.Subscription> response = Futures.immediateFuture(null);
     EasyMock.expect(pubsubRpcMock.get(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertNull(pubsub.getSubscription(SUBSCRIPTION));
   }
 
@@ -609,8 +639,8 @@ public void testGetSubscriptionAsync() throws ExecutionException, InterruptedExc
     Future<com.google.pubsub.v1.Subscription> response =
         Futures.immediateFuture(SUBSCRIPTION_INFO.toPb(PROJECT));
     EasyMock.expect(pubsubRpcMock.get(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     Subscription subscription = pubsub.getSubscriptionAsync(SUBSCRIPTION).get();
     assertEquals(
         new Subscription(pubsub, new SubscriptionInfo.BuilderImpl(COMPLETE_SUBSCRIPTION_INFO)),
@@ -623,8 +653,8 @@ public void testGetSubscriptionAsync_Null() throws ExecutionException, Interrupt
         GetSubscriptionRequest.newBuilder().setSubscription(SUBSCRIPTION_NAME_PB).build();
     Future<com.google.pubsub.v1.Subscription> response = Futures.immediateFuture(null);
     EasyMock.expect(pubsubRpcMock.get(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertNull(pubsub.getSubscriptionAsync(SUBSCRIPTION).get());
   }
 
@@ -635,8 +665,8 @@ public void testDeleteSubscription() {
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.delete(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertTrue(pubsub.deleteSubscription(SUBSCRIPTION));
   }
 
@@ -647,8 +677,8 @@ public void testDeleteSubscription_Null() {
         .build();
     Future<Empty> response = Futures.immediateFuture(null);
     EasyMock.expect(pubsubRpcMock.delete(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertFalse(pubsub.deleteSubscription(SUBSCRIPTION));
   }
 
@@ -659,8 +689,8 @@ public void testDeleteSubscriptionAsync() throws ExecutionException, Interrupted
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.delete(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertTrue(pubsub.deleteSubscriptionAsync(SUBSCRIPTION).get());
   }
 
@@ -671,8 +701,8 @@ public void testDeleteSubscriptionAsync_Null() throws ExecutionException, Interr
         .build();
     Future<Empty> response = Futures.immediateFuture(null);
     EasyMock.expect(pubsubRpcMock.delete(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     assertFalse(pubsub.deleteSubscriptionAsync(SUBSCRIPTION).get());
   }
 
@@ -684,8 +714,8 @@ public void testReplacePushConfig() {
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     pubsub.replacePushConfig(SUBSCRIPTION, PUSH_CONFIG);
   }
 
@@ -697,8 +727,8 @@ public void testReplacePushConfig_Null() {
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     pubsub.replacePushConfig(SUBSCRIPTION, null);
   }
 
@@ -710,8 +740,8 @@ public void testReplacePushConfigAsync() throws ExecutionException, InterruptedE
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     pubsub.replacePushConfigAsync(SUBSCRIPTION, PUSH_CONFIG).get();
   }
 
@@ -723,15 +753,16 @@ public void testReplacePushConfigAsync_Null() throws ExecutionException, Interru
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
-    pubsub = options.service();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub = new PubSubImpl(options, renewerMock);
     pubsub.replacePushConfigAsync(SUBSCRIPTION, null).get();
   }
 
   @Test
   public void testListSubscriptions() {
     String cursor = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(1);
     ListSubscriptionsRequest request = ListSubscriptionsRequest.newBuilder()
         .setProject(PROJECT_PB)
         .build();
@@ -744,7 +775,7 @@ public void testListSubscriptions() {
         .build();
     Future<ListSubscriptionsResponse> futureResponse = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Page<Subscription> page = pubsub.listSubscriptions();
     assertEquals(cursor, page.nextPageCursor());
     assertArrayEquals(subscriptionList.toArray(),
@@ -754,7 +785,8 @@ public void testListSubscriptions() {
   @Test
   public void testListSubscriptionsNextPage() {
     String cursor1 = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(2);
     ListSubscriptionsRequest request1 = ListSubscriptionsRequest.newBuilder()
         .setProject(PROJECT_PB)
         .build();
@@ -780,7 +812,7 @@ public void testListSubscriptionsNextPage() {
     Future<ListSubscriptionsResponse> futureResponse2 = Futures.immediateFuture(response2);
     EasyMock.expect(pubsubRpcMock.list(request1)).andReturn(futureResponse1);
     EasyMock.expect(pubsubRpcMock.list(request2)).andReturn(futureResponse2);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Page<Subscription> page = pubsub.listSubscriptions();
     assertEquals(cursor1, page.nextPageCursor());
     assertArrayEquals(subscriptionList1.toArray(),
@@ -793,7 +825,8 @@ public void testListSubscriptionsNextPage() {
 
   @Test
   public void testListSubscriptionsEmpty() {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(1);
     ListSubscriptionsRequest request = ListSubscriptionsRequest.newBuilder()
         .setProject(PROJECT_PB)
         .build();
@@ -804,7 +837,7 @@ public void testListSubscriptionsEmpty() {
         .build();
     Future<ListSubscriptionsResponse> futureResponse = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Page<Subscription> page = pubsub.listSubscriptions();
     assertNull(page.nextPageCursor());
     assertNull(page.nextPage());
@@ -815,7 +848,8 @@ public void testListSubscriptionsEmpty() {
   @Test
   public void testListSubscriptionsWithOptions() {
     String cursor = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(1);
     ListSubscriptionsRequest request = ListSubscriptionsRequest.newBuilder()
         .setProject(PROJECT_PB)
         .setPageSize(42)
@@ -830,7 +864,7 @@ public void testListSubscriptionsWithOptions() {
         .build();
     Future<ListSubscriptionsResponse> futureResponse = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Page<Subscription> page =
         pubsub.listSubscriptions(ListOption.pageSize(42), ListOption.pageToken(cursor));
     assertNull(page.nextPageCursor());
@@ -842,7 +876,8 @@ public void testListSubscriptionsWithOptions() {
   @Test
   public void testListSubscriptionsAsync() throws ExecutionException, InterruptedException {
     String cursor = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(1);
     ListSubscriptionsRequest request = ListSubscriptionsRequest.newBuilder()
         .setProject(PROJECT_PB)
         .build();
@@ -855,7 +890,7 @@ public void testListSubscriptionsAsync() throws ExecutionException, InterruptedE
         .build();
     Future<ListSubscriptionsResponse> futureResponse = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     AsyncPage<Subscription> page = pubsub.listSubscriptionsAsync().get();
     assertEquals(cursor, page.nextPageCursor());
     assertArrayEquals(subscriptionList.toArray(),
@@ -865,7 +900,8 @@ public void testListSubscriptionsAsync() throws ExecutionException, InterruptedE
   @Test
   public void testListSubscriptionsAsyncNextPage() throws ExecutionException, InterruptedException {
     String cursor1 = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(2);
     ListSubscriptionsRequest request1 = ListSubscriptionsRequest.newBuilder()
         .setProject(PROJECT_PB)
         .build();
@@ -891,7 +927,7 @@ public void testListSubscriptionsAsyncNextPage() throws ExecutionException, Inte
     Future<ListSubscriptionsResponse> futureResponse2 = Futures.immediateFuture(response2);
     EasyMock.expect(pubsubRpcMock.list(request1)).andReturn(futureResponse1);
     EasyMock.expect(pubsubRpcMock.list(request2)).andReturn(futureResponse2);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     AsyncPage<Subscription> page = pubsub.listSubscriptionsAsync().get();
     assertEquals(cursor1, page.nextPageCursor());
     assertArrayEquals(subscriptionList1.toArray(),
@@ -904,7 +940,8 @@ public void testListSubscriptionsAsyncNextPage() throws ExecutionException, Inte
 
   @Test
   public void testListSubscriptionsAsyncEmpty() throws ExecutionException, InterruptedException {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(1);
     ListSubscriptionsRequest request = ListSubscriptionsRequest.newBuilder()
         .setProject(PROJECT_PB)
         .build();
@@ -915,7 +952,7 @@ public void testListSubscriptionsAsyncEmpty() throws ExecutionException, Interru
         .build();
     Future<ListSubscriptionsResponse> futureResponse = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     AsyncPage<Subscription> page = pubsub.listSubscriptionsAsync().get();
     assertNull(page.nextPageCursor());
     assertNull(page.nextPageAsync().get());
@@ -928,7 +965,8 @@ public void testListSubscriptionsAsyncEmpty() throws ExecutionException, Interru
   public void testListSubscriptionsAsyncWithOptions()
       throws ExecutionException, InterruptedException {
     String cursor = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
+    resetOptionsForList(1);
     ListSubscriptionsRequest request = ListSubscriptionsRequest.newBuilder()
         .setProject(PROJECT_PB)
         .setPageSize(42)
@@ -943,7 +981,7 @@ public void testListSubscriptionsAsyncWithOptions()
         .build();
     Future<ListSubscriptionsResponse> futureResponse = Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     AsyncPage<Subscription> page =
         pubsub.listSubscriptionsAsync(ListOption.pageSize(42), ListOption.pageToken(cursor)).get();
     assertNull(page.nextPageCursor());
@@ -956,7 +994,7 @@ public void testListSubscriptionsAsyncWithOptions()
   @Test
   public void testListTopicSubscriptions() {
     String cursor = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ListTopicSubscriptionsRequest request = ListTopicSubscriptionsRequest.newBuilder()
         .setTopic(TOPIC_NAME_PB)
         .build();
@@ -970,7 +1008,7 @@ public void testListTopicSubscriptions() {
     Future<ListTopicSubscriptionsResponse> futureResponse =
         Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Page<SubscriptionId> page = pubsub.listSubscriptions(TOPIC);
     assertEquals(cursor, page.nextPageCursor());
     assertArrayEquals(subscriptionList.toArray(),
@@ -980,7 +1018,7 @@ public void testListTopicSubscriptions() {
   @Test
   public void testListTopicSubscriptionsNextPage() {
     String cursor1 = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ListTopicSubscriptionsRequest request1 = ListTopicSubscriptionsRequest.newBuilder()
         .setTopic(TOPIC_NAME_PB)
         .build();
@@ -1008,7 +1046,7 @@ public void testListTopicSubscriptionsNextPage() {
         Futures.immediateFuture(response2);
     EasyMock.expect(pubsubRpcMock.list(request1)).andReturn(futureResponse1);
     EasyMock.expect(pubsubRpcMock.list(request2)).andReturn(futureResponse2);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Page<SubscriptionId> page = pubsub.listSubscriptions(TOPIC);
     assertEquals(cursor1, page.nextPageCursor());
     assertArrayEquals(subscriptionList1.toArray(),
@@ -1021,7 +1059,7 @@ public void testListTopicSubscriptionsNextPage() {
 
   @Test
   public void testListTopicSubscriptionsEmpty() {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ListTopicSubscriptionsRequest request = ListTopicSubscriptionsRequest.newBuilder()
         .setTopic(TOPIC_NAME_PB)
         .build();
@@ -1033,7 +1071,7 @@ public void testListTopicSubscriptionsEmpty() {
     Future<ListTopicSubscriptionsResponse> futureResponse =
         Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Page<SubscriptionId> page = pubsub.listSubscriptions(TOPIC);
     assertNull(page.nextPageCursor());
     assertNull(page.nextPage());
@@ -1044,7 +1082,7 @@ public void testListTopicSubscriptionsEmpty() {
   @Test
   public void testListTopicSubscriptionsWithOptions() {
     String cursor = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ListTopicSubscriptionsRequest request = ListTopicSubscriptionsRequest.newBuilder()
         .setTopic(TOPIC_NAME_PB)
         .setPageSize(42)
@@ -1060,7 +1098,7 @@ public void testListTopicSubscriptionsWithOptions() {
     Future<ListTopicSubscriptionsResponse> futureResponse =
         Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Page<SubscriptionId> page =
         pubsub.listSubscriptions(TOPIC, ListOption.pageSize(42), ListOption.pageToken(cursor));
     assertNull(page.nextPageCursor());
@@ -1072,7 +1110,7 @@ public void testListTopicSubscriptionsWithOptions() {
   @Test
   public void testListTopicSubscriptionsAsync() throws ExecutionException, InterruptedException {
     String cursor = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ListTopicSubscriptionsRequest request = ListTopicSubscriptionsRequest.newBuilder()
         .setTopic(TOPIC_NAME_PB)
         .build();
@@ -1086,7 +1124,7 @@ public void testListTopicSubscriptionsAsync() throws ExecutionException, Interru
     Future<ListTopicSubscriptionsResponse> futureResponse =
         Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     AsyncPage<SubscriptionId> page = pubsub.listSubscriptionsAsync(TOPIC).get();
     assertEquals(cursor, page.nextPageCursor());
     assertArrayEquals(subscriptionList.toArray(),
@@ -1097,7 +1135,7 @@ public void testListTopicSubscriptionsAsync() throws ExecutionException, Interru
   public void testListTopicSubscriptionsAsyncNextPage()
       throws ExecutionException, InterruptedException {
     String cursor1 = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ListTopicSubscriptionsRequest request1 = ListTopicSubscriptionsRequest.newBuilder()
         .setTopic(TOPIC_NAME_PB)
         .build();
@@ -1125,7 +1163,7 @@ public void testListTopicSubscriptionsAsyncNextPage()
         Futures.immediateFuture(response2);
     EasyMock.expect(pubsubRpcMock.list(request1)).andReturn(futureResponse1);
     EasyMock.expect(pubsubRpcMock.list(request2)).andReturn(futureResponse2);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     AsyncPage<SubscriptionId> page = pubsub.listSubscriptionsAsync(TOPIC).get();
     assertEquals(cursor1, page.nextPageCursor());
     assertArrayEquals(subscriptionList1.toArray(),
@@ -1139,7 +1177,7 @@ public void testListTopicSubscriptionsAsyncNextPage()
   @Test
   public void testListTopicSubscriptionsAsyncEmpty()
       throws ExecutionException, InterruptedException {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ListTopicSubscriptionsRequest request = ListTopicSubscriptionsRequest.newBuilder()
         .setTopic(TOPIC_NAME_PB)
         .build();
@@ -1151,7 +1189,7 @@ public void testListTopicSubscriptionsAsyncEmpty()
     Future<ListTopicSubscriptionsResponse> futureResponse =
         Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     AsyncPage<SubscriptionId> page = pubsub.listSubscriptionsAsync(TOPIC).get();
     assertNull(page.nextPageCursor());
     assertNull(page.nextPage());
@@ -1164,7 +1202,7 @@ public void testListTopicSubscriptionsAsyncEmpty()
   public void testListTopicSubscriptionsAsyncWithOptions()
       throws ExecutionException, InterruptedException {
     String cursor = "cursor";
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ListTopicSubscriptionsRequest request = ListTopicSubscriptionsRequest.newBuilder()
         .setTopic(TOPIC_NAME_PB)
         .setPageSize(42)
@@ -1180,7 +1218,7 @@ public void testListTopicSubscriptionsAsyncWithOptions()
     Future<ListTopicSubscriptionsResponse> futureResponse =
         Futures.immediateFuture(response);
     EasyMock.expect(pubsubRpcMock.list(request)).andReturn(futureResponse);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     AsyncPage<SubscriptionId> page = pubsub.listSubscriptionsAsync(
         TOPIC, ListOption.pageSize(42), ListOption.pageToken(cursor)).get();
     assertNull(page.nextPageCursor());
@@ -1190,63 +1228,123 @@ public void testListTopicSubscriptionsAsyncWithOptions()
         Iterables.toArray(page.values(), SubscriptionId.class));
   }
 
+  @Test
+  public void testPullMessages() {
+    pubsub = new PubSubImpl(options, renewerMock);
+    PullRequest request = PullRequest.newBuilder()
+        .setSubscription(SUBSCRIPTION_NAME_PB)
+        .setMaxMessages(42)
+        .setReturnImmediately(true)
+        .build();
+    List<ReceivedMessage> messageList = ImmutableList.of(
+        ReceivedMessage.fromPb(pubsub, SUBSCRIPTION, MESSAGE_PB1),
+        ReceivedMessage.fromPb(pubsub, SUBSCRIPTION, MESSAGE_PB2));
+    PullResponse response = PullResponse.newBuilder()
+        .addReceivedMessages(MESSAGE_PB1)
+        .addReceivedMessages(MESSAGE_PB2)
+        .build();
+    EasyMock.expect(pubsubRpcMock.pull(request)).andReturn(Futures.immediateFuture(response));
+    renewerMock.add(SUBSCRIPTION, ImmutableList.of("ackId1", "ackId2"));
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    Iterator<ReceivedMessage> messageIterator = pubsub.pull(SUBSCRIPTION, 42);
+    EasyMock.reset(renewerMock);
+    for (ReceivedMessage message : messageList) {
+      renewerMock.remove(SUBSCRIPTION, message.ackId());
+      EasyMock.expectLastCall();
+    }
+    EasyMock.replay(renewerMock);
+    while (messageIterator.hasNext()) {
+      messageIterator.next();
+    }
+  }
+
+  @Test
+  public void testPullMessagesAsync() throws ExecutionException, InterruptedException {
+    pubsub = new PubSubImpl(options, renewerMock);
+    PullRequest request = PullRequest.newBuilder()
+        .setSubscription(SUBSCRIPTION_NAME_PB)
+        .setMaxMessages(42)
+        .setReturnImmediately(true)
+        .build();
+    List<ReceivedMessage> messageList = ImmutableList.of(
+        ReceivedMessage.fromPb(pubsub, SUBSCRIPTION, MESSAGE_PB1),
+        ReceivedMessage.fromPb(pubsub, SUBSCRIPTION, MESSAGE_PB2));
+    PullResponse response = PullResponse.newBuilder()
+        .addReceivedMessages(MESSAGE_PB1)
+        .addReceivedMessages(MESSAGE_PB2)
+        .build();
+    EasyMock.expect(pubsubRpcMock.pull(request)).andReturn(Futures.immediateFuture(response));
+    renewerMock.add(SUBSCRIPTION, ImmutableList.of("ackId1", "ackId2"));
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    Iterator<ReceivedMessage> messageIterator = pubsub.pullAsync(SUBSCRIPTION, 42).get();
+    EasyMock.reset(renewerMock);
+    for (ReceivedMessage message : messageList) {
+      renewerMock.remove(SUBSCRIPTION, message.ackId());
+      EasyMock.expectLastCall();
+    }
+    EasyMock.replay(renewerMock);
+    while (messageIterator.hasNext()) {
+      messageIterator.next();
+    }
+  }
+
   @Test
   public void testAckOneMessage() {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     AcknowledgeRequest request = AcknowledgeRequest.newBuilder()
         .setSubscription(SUBSCRIPTION_NAME_PB)
         .addAckIds("ackId")
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.acknowledge(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     pubsub.ack(SUBSCRIPTION, "ackId");
   }
 
   @Test
   public void testAckOneMessageAsync() throws ExecutionException, InterruptedException {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     AcknowledgeRequest request = AcknowledgeRequest.newBuilder()
         .setSubscription(SUBSCRIPTION_NAME_PB)
         .addAckIds("ackId")
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.acknowledge(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Future<Void> future = pubsub.ackAsync(SUBSCRIPTION, "ackId");
     assertNull(future.get());
   }
 
   @Test
   public void testAckMoreMessages() {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     AcknowledgeRequest request = AcknowledgeRequest.newBuilder()
         .setSubscription(SUBSCRIPTION_NAME_PB)
         .addAllAckIds(ImmutableList.of("ackId1", "ackId2"))
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.acknowledge(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     pubsub.ack(SUBSCRIPTION, "ackId1", "ackId2");
   }
 
   @Test
   public void testAckMoreMessagesAsync() throws ExecutionException, InterruptedException {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     AcknowledgeRequest request = AcknowledgeRequest.newBuilder()
         .setSubscription(SUBSCRIPTION_NAME_PB)
         .addAllAckIds(ImmutableList.of("ackId1", "ackId2"))
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.acknowledge(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Future<Void> future = pubsub.ackAsync(SUBSCRIPTION, "ackId1", "ackId2");
     assertNull(future.get());
   }
 
   @Test
   public void testAckMessageList() {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     List<String> ackIds = ImmutableList.of("ackId1", "ackId2");
     AcknowledgeRequest request = AcknowledgeRequest.newBuilder()
         .setSubscription(SUBSCRIPTION_NAME_PB)
@@ -1254,13 +1352,13 @@ public void testAckMessageList() {
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.acknowledge(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     pubsub.ack(SUBSCRIPTION, ackIds);
   }
 
   @Test
   public void testAckMessageListAsync() throws ExecutionException, InterruptedException {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     List<String> ackIds = ImmutableList.of("ackId1", "ackId2");
     AcknowledgeRequest request = AcknowledgeRequest.newBuilder()
         .setSubscription(SUBSCRIPTION_NAME_PB)
@@ -1268,14 +1366,14 @@ public void testAckMessageListAsync() throws ExecutionException, InterruptedExce
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.acknowledge(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Future<Void> future = pubsub.ackAsync(SUBSCRIPTION, ackIds);
     assertNull(future.get());
   }
 
   @Test
   public void testNackOneMessage() {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ModifyAckDeadlineRequest request = ModifyAckDeadlineRequest.newBuilder()
         .setAckDeadlineSeconds(0)
         .setSubscription(SUBSCRIPTION_NAME_PB)
@@ -1283,13 +1381,13 @@ public void testNackOneMessage() {
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     pubsub.nack(SUBSCRIPTION, "ackId");
   }
 
   @Test
   public void testNackOneMessageAsync() throws ExecutionException, InterruptedException {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ModifyAckDeadlineRequest request = ModifyAckDeadlineRequest.newBuilder()
         .setAckDeadlineSeconds(0)
         .setSubscription(SUBSCRIPTION_NAME_PB)
@@ -1297,14 +1395,14 @@ public void testNackOneMessageAsync() throws ExecutionException, InterruptedExce
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Future<Void> future = pubsub.nackAsync(SUBSCRIPTION, "ackId");
     assertNull(future.get());
   }
 
   @Test
   public void testNackMoreMessages() {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ModifyAckDeadlineRequest request = ModifyAckDeadlineRequest.newBuilder()
         .setAckDeadlineSeconds(0)
         .setSubscription(SUBSCRIPTION_NAME_PB)
@@ -1312,13 +1410,13 @@ public void testNackMoreMessages() {
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     pubsub.nack(SUBSCRIPTION, "ackId1", "ackId2");
   }
 
   @Test
   public void testNackMoreMessagesAsync() throws ExecutionException, InterruptedException {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ModifyAckDeadlineRequest request = ModifyAckDeadlineRequest.newBuilder()
         .setAckDeadlineSeconds(0)
         .setSubscription(SUBSCRIPTION_NAME_PB)
@@ -1326,14 +1424,14 @@ public void testNackMoreMessagesAsync() throws ExecutionException, InterruptedEx
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Future<Void> future = pubsub.nackAsync(SUBSCRIPTION, "ackId1", "ackId2");
     assertNull(future.get());
   }
 
   @Test
   public void testNackMessageList() {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     List<String> ackIds = ImmutableList.of("ackId1", "ackId2");
     ModifyAckDeadlineRequest request = ModifyAckDeadlineRequest.newBuilder()
         .setAckDeadlineSeconds(0)
@@ -1342,13 +1440,13 @@ public void testNackMessageList() {
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     pubsub.nack(SUBSCRIPTION, ackIds);
   }
 
   @Test
   public void testNackMessageListAsync() throws ExecutionException, InterruptedException {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     List<String> ackIds = ImmutableList.of("ackId1", "ackId2");
     ModifyAckDeadlineRequest request = ModifyAckDeadlineRequest.newBuilder()
         .setAckDeadlineSeconds(0)
@@ -1357,14 +1455,14 @@ public void testNackMessageListAsync() throws ExecutionException, InterruptedExc
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Future<Void> future = pubsub.nackAsync(SUBSCRIPTION, ackIds);
     assertNull(future.get());
   }
 
   @Test
   public void testModifyAckDeadlineOneMessage() {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ModifyAckDeadlineRequest request = ModifyAckDeadlineRequest.newBuilder()
         .setAckDeadlineSeconds(10)
         .setSubscription(SUBSCRIPTION_NAME_PB)
@@ -1372,14 +1470,14 @@ public void testModifyAckDeadlineOneMessage() {
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     pubsub.modifyAckDeadline(SUBSCRIPTION, 10, TimeUnit.SECONDS, "ackId");
   }
 
   @Test
   public void testModifyAckDeadlineOneMessageAsync()
       throws ExecutionException, InterruptedException {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ModifyAckDeadlineRequest request = ModifyAckDeadlineRequest.newBuilder()
         .setAckDeadlineSeconds(10)
         .setSubscription(SUBSCRIPTION_NAME_PB)
@@ -1387,7 +1485,7 @@ public void testModifyAckDeadlineOneMessageAsync()
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Future<Void> future =
         pubsub.modifyAckDeadlineAsync(SUBSCRIPTION, 10, TimeUnit.SECONDS, "ackId");
     assertNull(future.get());
@@ -1395,7 +1493,7 @@ public void testModifyAckDeadlineOneMessageAsync()
 
   @Test
   public void testModifyAckDeadlineMoreMessages() {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ModifyAckDeadlineRequest request = ModifyAckDeadlineRequest.newBuilder()
         .setAckDeadlineSeconds(10)
         .setSubscription(SUBSCRIPTION_NAME_PB)
@@ -1403,14 +1501,14 @@ public void testModifyAckDeadlineMoreMessages() {
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     pubsub.modifyAckDeadline(SUBSCRIPTION, 10, TimeUnit.SECONDS, "ackId1", "ackId2");
   }
 
   @Test
   public void testModifyAckDeadlineMoreMessagesAsync()
       throws ExecutionException, InterruptedException {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     ModifyAckDeadlineRequest request = ModifyAckDeadlineRequest.newBuilder()
         .setAckDeadlineSeconds(10)
         .setSubscription(SUBSCRIPTION_NAME_PB)
@@ -1418,7 +1516,7 @@ public void testModifyAckDeadlineMoreMessagesAsync()
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Future<Void> future =
         pubsub.modifyAckDeadlineAsync(SUBSCRIPTION, 10, TimeUnit.SECONDS, "ackId1", "ackId2");
     assertNull(future.get());
@@ -1426,7 +1524,7 @@ public void testModifyAckDeadlineMoreMessagesAsync()
 
   @Test
   public void testModifyAckDeadlineMessageList() {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     List<String> ackIds = ImmutableList.of("ackId1", "ackId2");
     ModifyAckDeadlineRequest request = ModifyAckDeadlineRequest.newBuilder()
         .setAckDeadlineSeconds(10)
@@ -1435,14 +1533,14 @@ public void testModifyAckDeadlineMessageList() {
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     pubsub.modifyAckDeadline(SUBSCRIPTION, 10, TimeUnit.SECONDS, ackIds);
   }
 
   @Test
   public void testModifyAckDeadlineMessageListAsync()
       throws ExecutionException, InterruptedException {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     List<String> ackIds = ImmutableList.of("ackId1", "ackId2");
     ModifyAckDeadlineRequest request = ModifyAckDeadlineRequest.newBuilder()
         .setAckDeadlineSeconds(10)
@@ -1451,17 +1549,22 @@ public void testModifyAckDeadlineMessageListAsync()
         .build();
     Future<Empty> response = Futures.immediateFuture(Empty.getDefaultInstance());
     EasyMock.expect(pubsubRpcMock.modify(request)).andReturn(response);
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.replay(pubsubRpcMock, renewerMock);
     Future<Void> future = pubsub.modifyAckDeadlineAsync(SUBSCRIPTION, 10, TimeUnit.SECONDS, ackIds);
     assertNull(future.get());
   }
 
   @Test
   public void testClose() throws Exception {
-    pubsub = options.service();
+    pubsub = new PubSubImpl(options, renewerMock);
     pubsubRpcMock.close();
     EasyMock.expectLastCall();
-    EasyMock.replay(pubsubRpcMock);
+    EasyMock.expectLastCall();
+    renewerMock.close();
+    EasyMock.expectLastCall();
+    EasyMock.replay(pubsubRpcMock, renewerMock);
+    pubsub.close();
+    // closing again should do nothing
     pubsub.close();
   }
 }