pubsub: reject expired and duplicate messages

[pongad]

CC @csainty

This should let us clear through backlogs more quickly.

[csainty]

Generally like it, nice clean approach 👍

Won't handle a 0 extension period though will it? Since totalExpiration will be set to receive time

final Subscriber subscriber = Subscriber.newBuilder(subscriptionName, (msg, ack) -> {
    // do something
  })
  .setMaxAckExtensionPeriod(Duration.ZERO)
  .build();

[pongad]

That's right. This won't handle the case where max ack extension is zero. I'm a little concerned by this; this essentially changes the behavior from "we won't extend these" to "we'll never work on these". I'll make a small fix for this.

[csainty]

👍
I pulled this morning and ran my sample project with the build, hits exactly-once consumption. Thanks for looking in to it.

I think this will fix #3383 and #2465
Also make it easier to configure for https://github.com/GoogleCloudPlatform/google-cloud-java/issues/3152 by tweaking the max deadline so all the messages don't pull to a single subscriber and stay there, as currently reducing the max deadline to prevent this gets you in to double processing sadness.

[pongad]

Don't merge this yet. Someone from pubsub should take a look.