Recreation of GRPC publisher/subscribers

kafka-connector/src/main/java/com/google/pubsub/kafka/sink/CloudPubSubGRPCPublisher.java

@@ -22,25 +22,44 @@
 import com.google.pubsub.v1.PublisherGrpc;
 import com.google.pubsub.v1.PublisherGrpc.PublisherFutureStub;
 import java.io.IOException;
+import java.util.Random;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * A {@link CloudPubSubPublisher} that uses <a href="http://www.grpc.io/">gRPC</a> to send messages
- * to <a href="https://cloud.google.com/pubsub">Google Cloud Pub/Sub</a>.
+ * to <a href="https://cloud.google.com/pubsub">Google Cloud Pub/Sub</a>. This class is not thread-
+ * safe.
  */
 public class CloudPubSubGRPCPublisher implements CloudPubSubPublisher {
 
+  private static final Logger log = LoggerFactory.getLogger(CloudPubSubGRPCPublisher.class);
+  private long nextPublisherResetTime = 0;
   private PublisherFutureStub publisher;
+  private Random rand = new Random(System.currentTimeMillis());
 
   public CloudPubSubGRPCPublisher() {
-    try {
-      publisher = PublisherGrpc.newFutureStub(ConnectorUtils.getChannel());
-    } catch (IOException e) {
-      throw new RuntimeException("Could not create publisher stub; no publishes can occur.");
-    }
+    makePublisher();
   }
 
   @Override
   public ListenableFuture<PublishResponse> publish(PublishRequest request) {
+    if (System.currentTimeMillis() >= nextPublisherResetTime) {
+      makePublisher();
+    }
+
     return publisher.publish(request);
   }
+
+  private void makePublisher() {
+    try {
+      log.info("Creating publisher.");
+      publisher = PublisherGrpc.newFutureStub(ConnectorUtils.getChannel());
+      // We change the publisher every 25 - 35 minutes in order to avoid GOAWAY errors.
+      nextPublisherResetTime =
+          System.currentTimeMillis() + rand.nextInt(10 * 60 * 1000) + 25 * 60 * 1000;
+    } catch (IOException e) {
+      throw new RuntimeException("Could not create publisher stub; no publishes can occur.", e);
+    }
+  }
 }

kafka-connector/src/main/java/com/google/pubsub/kafka/source/CloudPubSubGRPCSubscriber.java

@@ -24,28 +24,49 @@
 import com.google.pubsub.v1.SubscriberGrpc;
 import com.google.pubsub.v1.SubscriberGrpc.SubscriberFutureStub;
 import java.io.IOException;
+import java.util.Random;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * A {@link CloudPubSubSubscriber} that uses <a href="http://www.grpc.io/">gRPC</a> to pull messages
- * from <a href="https://cloud.google.com/pubsub">Google Cloud Pub/Sub</a>.
+ * from <a href="https://cloud.google.com/pubsub">Google Cloud Pub/Sub</a>. This class is not
+ * thread-safe.
  */
 public class CloudPubSubGRPCSubscriber implements CloudPubSubSubscriber {
 
+  private static final Logger log = LoggerFactory.getLogger(CloudPubSubGRPCSubscriber.class);
+  private long nextSubscriberResetTime = 0;
   private SubscriberFutureStub subscriber;
+  private Random rand = new Random(System.currentTimeMillis());
 
   CloudPubSubGRPCSubscriber() {
-    try {
-      subscriber = SubscriberGrpc.newFutureStub(ConnectorUtils.getChannel());
-    } catch (IOException e) {
-      throw new RuntimeException("Could not create subscriber stub; no pulls can occur.");
-    }
+    makeSubscriber();
   }
 
   public ListenableFuture<PullResponse> pull(PullRequest request) {
+    if (System.currentTimeMillis() > nextSubscriberResetTime) {
+      makeSubscriber();
+    }
     return subscriber.pull(request);
   }
 
   public ListenableFuture<Empty> ackMessages(AcknowledgeRequest request) {
+    if (System.currentTimeMillis() > nextSubscriberResetTime) {
+      makeSubscriber();
+    }
     return subscriber.acknowledge(request);
   }
+
+  private void makeSubscriber() {
+    try {
+      log.info("Creating subscriber.");
+      subscriber = SubscriberGrpc.newFutureStub(ConnectorUtils.getChannel());
+      // We change the subscriber every 25 - 35 minutes in order to avoid GOAWAY errors.
+      nextSubscriberResetTime =
+          System.currentTimeMillis() + rand.nextInt(10 * 60 * 1000) + 25 * 60 * 1000;
+    } catch (IOException e) {
+      throw new RuntimeException("Could not create subscriber stub; no subscribing can occur.", e);
+    }
+  }
 }