---

charlesliqlogic · chingor13 · commit 364dc707817a · 2019-01-02T09:51:26.000-08:00
yaml --- r: 15031 b: refs/heads/autosynth-bigquerydatatransfer c: 6e2c2dd h: refs/heads/master i: 15029: bf84084 15027: 12130f0 15023: 5fd1d2a
diff --git a/[refs] b/[refs]
@@ -124,7 +124,7 @@ refs/tags/v0.68.0: 9cc799fcf68c82ab431d425fefa58ef615ce8e5b
 refs/tags/v0.69.0: 78f67a29e8b9c46ba01de566a2eae0fd1c03edea
 refs/heads/autosynth-asset: cd8251de8c40e239ad24dcf9ed93ea2708a3eed5
 refs/heads/autosynth-automl: cced2f56bbef0499609073edbca6253e1df5e535
-refs/heads/autosynth-bigquerydatatransfer: fb2b690c57acdd37c2ac957788155606b28d5d9f
+refs/heads/autosynth-bigquerydatatransfer: 6e2c2dd890fe445cc896f0ddcb30309213759ee2
 refs/heads/autosynth-bigquerystorage: 99aee05df348f39d98b6fb23c292006f1d2a6c28
 refs/heads/autosynth-bigtable: fa0d1de9e264d7ecac8a3abc3de7a8364cfaf427
 refs/heads/autosynth-bigtable-admin: 6379a2bc712f2736c83de0e009b4d26da4fa82ca
diff --git a/branches/autosynth-bigquerydatatransfer/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/AckReplyConsumer.java b/branches/autosynth-bigquerydatatransfer/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/AckReplyConsumer.java
@@ -29,4 +29,6 @@ public interface AckReplyConsumer {
    * message.
    */
   void nack();
+
+  void abandon();
 }
diff --git a/branches/autosynth-bigquerydatatransfer/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java b/branches/autosynth-bigquerydatatransfer/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java
@@ -133,6 +133,7 @@ private class AckHandler implements ApiFutureCallback<AckReply> {
     private final int outstandingBytes;
     private final long receivedTimeMillis;
     private final Instant totalExpiration;
+    private boolean extending = true;
 
     AckHandler(String ackId, int outstandingBytes, Instant totalExpiration) {
       this.ackId = ackId;
@@ -151,6 +152,7 @@ private void forget() {
          */
         return;
       }
+      extending = false;
       flowController.release(1, outstandingBytes);
       messagesWaiter.incrementPendingMessages(-1);
       processOutstandingBatches();
@@ -417,6 +419,11 @@ public void ack() {
             public void nack() {
               response.set(AckReply.NACK);
             }
+
+            @Override
+            public void abandon() {
+              ackHandler.forget();
+            }
           };
       ApiFutures.addCallback(response, ackHandler, MoreExecutors.directExecutor());
       executor.execute(
@@ -471,6 +478,9 @@ void extendDeadlines() {
     Instant extendTo = now.plusSeconds(extendSeconds);
 
     for (Map.Entry<String, AckHandler> entry : pendingMessages.entrySet()) {
+      if (!entry.getValue().extending) {
+        continue;
+      }
       String ackId = entry.getKey();
       Instant totalExpiration = entry.getValue().totalExpiration;
       if (totalExpiration.isAfter(extendTo)) {
diff --git a/branches/autosynth-bigquerydatatransfer/google-cloud-clients/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/MessageDispatcherTest.java b/branches/autosynth-bigquerydatatransfer/google-cloud-clients/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/MessageDispatcherTest.java
@@ -148,6 +148,14 @@ public void testNack() throws Exception {
     assertThat(sentModAcks).contains(ModAckItem.of(TEST_MESSAGE.getAckId(), 0));
   }
 
+  @Test
+  public void testAbandon() throws Exception {
+    dispatcher.processReceivedMessages(Collections.singletonList(TEST_MESSAGE), NOOP_RUNNABLE);
+    consumers.take().abandon();
+    dispatcher.extendDeadlines();
+    assertThat(sentModAcks).doesNotContain(TEST_MESSAGE.getAckId());
+  }
+
   @Test
   public void testExtension() throws Exception {
     dispatcher.processReceivedMessages(Collections.singletonList(TEST_MESSAGE), NOOP_RUNNABLE);

Original file line number	Diff line number	Diff line change
`@@ -29,4 +29,6 @@ public interface AckReplyConsumer {`
`29`	`29`	`* message.`
`30`	`30`	`*/`
`31`	`31`	`void nack();`
	`32`	`+`
	`33`	`+ void abandon();`
`32`	`34`	`}`