diff --git a/pubsub/internal/longtest/endtoend_test.go b/pubsub/internal/longtest/endtoend_test.go
index fb4be40..dd1d362 100644
--- a/pubsub/internal/longtest/endtoend_test.go
+++ b/pubsub/internal/longtest/endtoend_test.go
@@ -15,12 +15,10 @@
 package longtest
 
 import (
-	"bytes"
 	"context"
 	"fmt"
 	"log"
 	"math/rand"
-	"os"
 	"sync"
 	"testing"
 	"time"
@@ -28,6 +26,8 @@
 	"cloud.google.com/go/internal/testutil"
 	"cloud.google.com/go/pubsub"
 	"google.golang.org/api/option"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 const (
@@ -38,39 +38,18 @@
 	numAcceptableDups       = int(nMessages * acceptableDupPercentage / 100)
 )
 
-// Buffer log messages to debug failures.
-var logBuf bytes.Buffer
-
 // The end-to-end pumps many messages into a topic and tests that they are all
 // delivered to each subscription for the topic. It also tests that messages
 // are not unexpectedly redelivered.
-func TestIntegration_EndToEnd(t *testing.T) {
-	if testing.Short() {
-		t.Skip("Integration tests skipped in short mode")
-	}
-	log.SetOutput(&logBuf)
-	ctx := context.Background()
-	ts := testutil.TokenSource(ctx, pubsub.ScopePubSub, pubsub.ScopeCloudPlatform)
-	if ts == nil {
-		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
-	}
-
-	now := time.Now()
-	topicName := fmt.Sprintf("endtoend-%d", now.UnixNano())
-	subPrefix := fmt.Sprintf("endtoend-%d", now.UnixNano())
-
-	client, err := pubsub.NewClient(ctx, testutil.ProjID(), option.WithTokenSource(ts))
-	if err != nil {
-		t.Fatalf("Creating client error: %v", err)
-	}
-
-	var topic *pubsub.Topic
-	if topic, err = client.CreateTopic(ctx, topicName); err != nil {
-		t.Fatalf("CreateTopic error: %v", err)
-	}
-	defer topic.Delete(ctx)
+func TestEndToEnd_Dupes(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	client, topic, cleanup := prepareEndToEndTest(ctx, t)
+	defer cleanup()
+	subPrefix := fmt.Sprintf("endtoend-%d", time.Now().UnixNano())
 
 	// Two subscriptions to the same topic.
+	var err error
 	var subs [2]*pubsub.Subscription
 	for i := 0; i < len(subs); i++ {
 		subs[i], err = client.CreateSubscription(ctx, fmt.Sprintf("%s-%d", subPrefix, i), pubsub.SubscriptionConfig{
@@ -108,7 +87,7 @@
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			con.consume(cctx, t, sub)
+			con.consume(ctx, t, sub)
 		}()
 	}
 	// Wait for a while after the last message before declaring quiescence.
@@ -142,7 +121,6 @@
 		}
 	}
 	wg.Wait()
-	ok := true
 	for i, con := range consumers {
 		var numDups int
 		var zeroes int
@@ -155,14 +133,92 @@
 
 		if zeroes > 0 {
 			t.Errorf("Consumer %d: %d messages never arrived", i, zeroes)
-			ok = false
 		} else if numDups > numAcceptableDups {
 			t.Errorf("Consumer %d: Willing to accept %d dups (%v duplicated of %d messages), but got %d", i, numAcceptableDups, acceptableDupPercentage, int(nMessages), numDups)
-			ok = false
 		}
 	}
-	if !ok {
-		logBuf.WriteTo(os.Stdout)
+}
+
+func TestEndToEnd_LongProcessingTime(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	client, topic, cleanup := prepareEndToEndTest(ctx, t)
+	defer cleanup()
+	subPrefix := fmt.Sprintf("endtoend-%d", time.Now().UnixNano())
+
+	// Two subscriptions to the same topic.
+	sub, err := client.CreateSubscription(ctx, subPrefix+"-00", pubsub.SubscriptionConfig{
+		Topic:       topic,
+		AckDeadline: ackDeadline,
+	})
+	if err != nil {
+		t.Fatalf("CreateSub error: %v", err)
+	}
+	defer sub.Delete(ctx)
+
+	// Tests the issue found in https://github.com/googleapis/google-cloud-go/issues/1247.
+	sub.ReceiveSettings.Synchronous = true
+	sub.ReceiveSettings.MaxOutstandingMessages = 500
+
+	err = publish(ctx, topic, 1000)
+	topic.Stop()
+	if err != nil {
+		t.Fatalf("publish: %v", err)
+	}
+
+	// recv provides an indication that messages are still arriving.
+	recv := make(chan struct{})
+	consumer := consumer{
+		counts:    make(map[string]int),
+		recv:      recv,
+		durations: []time.Duration{time.Hour},
+		processingDelay: func() time.Duration {
+			return time.Duration(1+rand.Int63n(120)) * time.Second
+		},
+	}
+	go consumer.consume(ctx, t, sub)
+	// Wait for a while after the last message before declaring quiescence.
+	// We wait a multiple of the ack deadline, for two reasons:
+	// 1. To detect if messages are redelivered after having their ack
+	//    deadline extended.
+	// 2. To wait for redelivery of messages that were en route when a Receive
+	//    is canceled. This can take considerably longer than the ack deadline.
+	quiescenceDur := ackDeadline * 6
+	quiescenceTimer := time.NewTimer(quiescenceDur)
+loop:
+	for {
+		select {
+		case <-recv:
+			// Reset timer so we wait quiescenceDur after the last message.
+			// See https://godoc.org/time#Timer.Reset for why the Stop
+			// and channel drain are necessary.
+			if !quiescenceTimer.Stop() {
+				<-quiescenceTimer.C
+			}
+			quiescenceTimer.Reset(quiescenceDur)
+
+		case <-quiescenceTimer.C:
+			cancel()
+			log.Println("quiesced")
+			break loop
+
+		case <-ctx.Done():
+			t.Fatal("timed out")
+		}
+	}
+	var numDups int
+	var zeroes int
+	for _, v := range consumer.counts {
+		if v == 0 {
+			zeroes++
+		}
+		numDups += v - 1
+	}
+
+	if zeroes > 0 {
+		t.Errorf("%d messages never arrived", zeroes)
+	} else if numDups > numAcceptableDups {
+		t.Errorf("Willing to accept %d dups (%v duplicated of %d messages), but got %d", numAcceptableDups, acceptableDupPercentage, int(nMessages), numDups)
 	}
 }
 
@@ -184,14 +240,20 @@
 
 // consumer consumes messages according to its configuration.
 type consumer struct {
+	// A consumer will spin out a Receive for each duration, which will be
+	// canceled after each duration and the next one spun up. For example, if
+	// there are 5 3 second durations, then there will be 5 3 second Receives.
 	durations []time.Duration
 
 	// A value is sent to recv each time Inc is called.
 	recv chan struct{}
 
-	mu     sync.Mutex
-	counts map[string]int
-	total  int
+	// How long to wait for before acking.
+	processingDelay func() time.Duration
+
+	mu         sync.Mutex
+	counts     map[string]int // msgID: recvdAmt
+	totalRecvd int
 }
 
 // consume reads messages from a subscription, and keeps track of what it receives in mc.
@@ -201,13 +263,12 @@
 		ctx2, cancel := context.WithTimeout(ctx, dur)
 		defer cancel()
 		id := sub.String()[len(sub.String())-1:]
-		log.Printf("%s: start receive", id)
-		prev := c.total
+		t.Logf("%s: start receive", id)
+		prev := c.totalRecvd
 		err := sub.Receive(ctx2, c.process)
-		log.Printf("%s: end receive; read %d", id, c.total-prev)
-		if err != nil {
-			t.Errorf("error from Receive: %v", err)
-			return
+		t.Logf("%s: end receive; read %d", id, c.totalRecvd-prev)
+		if serr, _ := status.FromError(err); serr.Code() != codes.Canceled {
+			panic(err)
 		}
 		select {
 		case <-ctx.Done():
@@ -215,17 +276,56 @@
 		default:
 		}
 	}
+	return
 }
 
 // process handles a message and records it in mc.
 func (c *consumer) process(_ context.Context, m *pubsub.Message) {
 	c.mu.Lock()
 	c.counts[m.ID]++
-	c.total++
+	c.totalRecvd++
 	c.mu.Unlock()
 	c.recv <- struct{}{}
+
+	var delay time.Duration
+	if c.processingDelay == nil {
+		delay = time.Duration(rand.Intn(int(ackDeadline * 3)))
+	} else {
+		delay = c.processingDelay()
+	}
+
 	// Simulate time taken to process m, while continuing to process more messages.
 	// Some messages will need to have their ack deadline extended due to this delay.
-	delay := rand.Intn(int(ackDeadline * 3))
-	time.AfterFunc(time.Duration(delay), m.Ack)
+	time.AfterFunc(delay, func() {
+		m.Ack()
+	})
+}
+
+// Remember to call cleanup!
+func prepareEndToEndTest(ctx context.Context, t *testing.T) (*pubsub.Client, *pubsub.Topic, func()) {
+	if testing.Short() {
+		t.Skip("Integration tests skipped in short mode")
+	}
+	ts := testutil.TokenSource(ctx, pubsub.ScopePubSub, pubsub.ScopeCloudPlatform)
+	if ts == nil {
+		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
+	}
+
+	now := time.Now()
+	topicName := fmt.Sprintf("endtoend-%d", now.UnixNano())
+
+	client, err := pubsub.NewClient(ctx, testutil.ProjID(), option.WithTokenSource(ts))
+	if err != nil {
+		t.Fatalf("Creating client error: %v", err)
+	}
+
+	var topic *pubsub.Topic
+	if topic, err = client.CreateTopic(ctx, topicName); err != nil {
+		t.Fatalf("CreateTopic error: %v", err)
+	}
+
+	return client, topic, func() {
+		topic.Delete(ctx)
+		client.Close()
+	}
 }
diff --git a/pubsub/iterator.go b/pubsub/iterator.go
index 5b3415e..0d2a715 100644
--- a/pubsub/iterator.go
+++ b/pubsub/iterator.go
@@ -214,7 +214,7 @@
 		it.keepAliveDeadlines[m.ackID] = maxExt
 		// Don't change the mod-ack if the message is going to be nacked. This is
 		// possible if there are retries.
-		if !it.pendingNacks[m.ackID] && !it.po.synchronous {
+		if !it.pendingNacks[m.ackID] {
 			ackIDs[m.ackID] = true
 		}
 	}