---

sduskis · web-flow · commit 4b34b466d571 · 2019-04-15T14:26:14.000-04:00
yaml --- r: 30677 b: refs/heads/autosynth-bigquerydatatransfer c: 2fc4e55 h: refs/heads/master i: 30675: 5f5875a
diff --git a/[refs] b/[refs]
@@ -122,7 +122,7 @@ refs/tags/v0.68.0: 9cc799fcf68c82ab431d425fefa58ef615ce8e5b
 refs/tags/v0.69.0: 78f67a29e8b9c46ba01de566a2eae0fd1c03edea
 refs/heads/autosynth-asset: bdb45634a0fe8f7a510692b56b31f5312e25f453
 refs/heads/autosynth-automl: 22f9dd5b6f5df8dbfa7da0126864d565229519b2
-refs/heads/autosynth-bigquerydatatransfer: 78360de040eb4d86d1b25a7792269d2952174d66
+refs/heads/autosynth-bigquerydatatransfer: 2fc4e551b5263fe1e902efdbc1fd01acf8e3b5e9
 refs/heads/autosynth-bigquerystorage: d2c53da3b012e38c662e4df0738042435f19365f
 refs/heads/autosynth-bigtable: 9e5429f45cf9face9fed585d0233534993e36b58
 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/MessageDispatcher.java b/branches/autosynth-bigquerydatatransfer/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java
@@ -104,10 +104,6 @@ private PendingModifyAckDeadline(int deadlineExtensionSeconds, Collection<String
       this.deadlineExtensionSeconds = deadlineExtensionSeconds;
     }
 
-    public void addAckId(String ackId) {
-      ackIds.add(ackId);
-    }
-
     @Override
     public String toString() {
       return String.format(
@@ -129,7 +125,7 @@ private class AckHandler implements ApiFutureCallback<AckReply> {
     private final long receivedTimeMillis;
     private final Instant totalExpiration;
 
-    AckHandler(String ackId, int outstandingBytes, Instant totalExpiration) {
+    private AckHandler(String ackId, int outstandingBytes, Instant totalExpiration) {
       this.ackId = ackId;
       this.outstandingBytes = outstandingBytes;
       this.receivedTimeMillis = clock.millisTime();
@@ -182,7 +178,7 @@ public void onSuccess(AckReply reply) {
     }
   }
 
-  public interface AckProcessor {
+  interface AckProcessor {
     void sendAckOperations(
         List<String> acksToSend, List<PendingModifyAckDeadline> ackDeadlineExtensions);
   }
@@ -211,7 +207,7 @@ void sendAckOperations(
     this.clock = clock;
   }
 
-  public void start() {
+  void start() {
     final Runnable setExtendDeadline =
         new Runnable() {
           @Override
@@ -264,7 +260,7 @@ public void run() {
     }
   }
 
-  public void stop() {
+  void stop() {
     messagesWaiter.waitNoMessages();
     jobLock.lock();
     try {
@@ -288,17 +284,17 @@ int getMessageDeadlineSeconds() {
     return messageDeadlineSeconds.get();
   }
 
-  static class OutstandingMessage {
+  private static class OutstandingMessage {
     private final ReceivedMessage receivedMessage;
     private final AckHandler ackHandler;
 
-    public OutstandingMessage(ReceivedMessage receivedMessage, AckHandler ackHandler) {
+    private OutstandingMessage(ReceivedMessage receivedMessage, AckHandler ackHandler) {
       this.receivedMessage = receivedMessage;
       this.ackHandler = ackHandler;
     }
   }
 
-  public void processReceivedMessages(List<ReceivedMessage> messages) {
+  void processReceivedMessages(List<ReceivedMessage> messages) {
     Instant totalExpiration = now().plus(maxAckExtensionPeriod);
     List<OutstandingMessage> outstandingBatch = new ArrayList<>(messages.size());
     for (ReceivedMessage message : messages) {
diff --git a/branches/autosynth-bigquerydatatransfer/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageWaiter.java b/branches/autosynth-bigquerydatatransfer/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageWaiter.java
@@ -17,7 +17,6 @@
 package com.google.cloud.pubsub.v1;
 
 import com.google.api.core.InternalApi;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 /** A barrier kind of object that helps to keep track and synchronously wait on pending messages. */
 class MessageWaiter {
@@ -35,16 +34,10 @@ public synchronized void incrementPendingMessages(int messages) {
   }
 
   public synchronized void waitNoMessages() {
-    waitNoMessages(new AtomicBoolean());
-  }
-
-  @InternalApi
-  synchronized void waitNoMessages(AtomicBoolean waitReached) {
     boolean interrupted = false;
     try {
       while (pendingMessages > 0) {
         try {
-          waitReached.set(true);
           wait();
         } catch (InterruptedException e) {
           // Ignored, uninterruptibly.
diff --git a/branches/autosynth-bigquerydatatransfer/google-cloud-clients/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/MessageWaiterTest.java b/branches/autosynth-bigquerydatatransfer/google-cloud-clients/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/MessageWaiterTest.java
@@ -18,7 +18,6 @@
 
 import static org.junit.Assert.assertEquals;
 
-import java.util.concurrent.atomic.AtomicBoolean;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
@@ -32,22 +31,21 @@ public void test() throws Exception {
     final MessageWaiter waiter = new MessageWaiter();
     waiter.incrementPendingMessages(1);
 
-    final AtomicBoolean waitReached = new AtomicBoolean();
-
+    final Thread mainThread = Thread.currentThread();
     Thread t =
         new Thread(
             new Runnable() {
               @Override
               public void run() {
-                while (!waitReached.get()) {
+                while (mainThread.getState() != Thread.State.WAITING) {
                   Thread.yield();
                 }
                 waiter.incrementPendingMessages(-1);
               }
             });
     t.start();
 
-    waiter.waitNoMessages(waitReached);
+    waiter.waitNoMessages();
     t.join();
 
     assertEquals(0, waiter.pendingMessages());
