fix: if a flow control setting is not provided use zero

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java

@@ -216,8 +216,10 @@ private void initialize() {
             .setSubscription(subscription)
             .setStreamAckDeadlineSeconds(60)
             .setClientId(clientId)
-            .setMaxOutstandingMessages(flowControlSettings.getMaxOutstandingElementCount())
-            .setMaxOutstandingBytes(flowControlSettings.getMaxOutstandingRequestBytes())
+            .setMaxOutstandingMessages(
+                valueOrZero(flowControlSettings.getMaxOutstandingElementCount()))
+            .setMaxOutstandingBytes(
+                valueOrZero(flowControlSettings.getMaxOutstandingRequestBytes()))
             .build());
 
     /**
@@ -281,6 +283,10 @@ public void run() {
         MoreExecutors.directExecutor());
   }
 
+  private Long valueOrZero(Long value) {
+    return value != null ? value : 0;
+  }
+
   private boolean isAlive() {
     State state = state(); // Read the state only once.
     return state == State.RUNNING || state == State.STARTING;

google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/SubscriberTest.java

@@ -19,6 +19,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
+import com.google.api.gax.batching.FlowControlSettings;
 import com.google.api.gax.core.ExecutorProvider;
 import com.google.api.gax.core.FixedExecutorProvider;
 import com.google.api.gax.core.InstantiatingExecutorProvider;
@@ -238,6 +239,8 @@ private Builder getTestSubscriberBuilder(MessageReceiver receiver) {
         .setCredentialsProvider(NoCredentialsProvider.create())
         .setClock(fakeExecutor.getClock())
         .setParallelPullCount(1)
-        .setMaxDurationPerAckExtension(Duration.ofSeconds(5));
+        .setMaxDurationPerAckExtension(Duration.ofSeconds(5))
+        .setFlowControlSettings(
+            FlowControlSettings.newBuilder().setMaxOutstandingElementCount(1000L).build());
   }
 }