test(pubsub): fix JSpecify compatibility issues on Java 8 for Mockito tests (#13808)

Update Mockito mocks in pubsub tests to use
`withSettings().withoutAnnotations()` to avoid `ArrayStoreException` on
Java 8 due to JSpecify's `@NullMarked` annotation. Explicitly mock
callables instead of relying on `RETURNS_DEEP_STUBS` which does not
propagate the settings.

Author: lqiu96

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

@@ -94,7 +94,7 @@ public class MessageDispatcherTest {
   public void setUp() {
     systemExecutor = new FakeScheduledExecutorService();
     clock = new FakeClock();
-    mockAckLatencyDistribution = mock(Distribution.class);
+    mockAckLatencyDistribution = mock(Distribution.class, withSettings().withoutAnnotations());
 
     mockAckProcessor = mock(MessageDispatcher.AckProcessor.class);
     messageContainsDeliveryAttempt = true;
@@ -710,8 +710,9 @@ private MessageDispatcher getMessageDispatcherFromBuilder(
             .setMinDurationPerAckExtensionDefaultUsed(true)
             .setMaxDurationPerAckExtension(Subscriber.DEFAULT_MAX_ACK_DEADLINE_EXTENSION)
             .setMaxDurationPerAckExtensionDefaultUsed(true)
-            .setAckLatencyDistribution(mock(Distribution.class))
-            .setFlowController(mock(FlowController.class))
+            .setAckLatencyDistribution(
+                mock(Distribution.class, withSettings().withoutAnnotations()))
+            .setFlowController(mock(FlowController.class, withSettings().withoutAnnotations()))
             .setExecutor(executor)
             .setSubscriptionName(MOCK_SUBSCRIPTION_NAME)
             .setSystemExecutor(systemExecutor)
@@ -734,8 +735,9 @@ private MessageDispatcher getMessageDispatcherFromBuilder(
             .setMinDurationPerAckExtensionDefaultUsed(true)
             .setMaxDurationPerAckExtension(Subscriber.DEFAULT_MAX_ACK_DEADLINE_EXTENSION)
             .setMaxDurationPerAckExtensionDefaultUsed(true)
-            .setAckLatencyDistribution(mock(Distribution.class))
-            .setFlowController(mock(FlowController.class))
+            .setAckLatencyDistribution(
+                mock(Distribution.class, withSettings().withoutAnnotations()))
+            .setFlowController(mock(FlowController.class, withSettings().withoutAnnotations()))
             .setExecutor(MoreExecutors.newDirectExecutorService())
             .setSubscriptionName(MOCK_SUBSCRIPTION_NAME)
             .setSystemExecutor(systemExecutor)

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

@@ -33,9 +33,11 @@
 import com.google.api.gax.rpc.ResponseObserver;
 import com.google.api.gax.rpc.StatusCode;
 import com.google.api.gax.rpc.StreamController;
+import com.google.api.gax.rpc.UnaryCallable;
 import com.google.cloud.pubsub.v1.stub.SubscriberStub;
 import com.google.common.collect.Lists;
 import com.google.protobuf.Any;
+import com.google.protobuf.Empty;
 import com.google.pubsub.v1.AcknowledgeRequest;
 import com.google.pubsub.v1.ModifyAckDeadlineRequest;
 import com.google.pubsub.v1.StreamingPullRequest;
@@ -66,6 +68,9 @@ public class StreamingSubscriberConnectionTest {
   private FakeScheduledExecutorService executor;
   private FakeClock clock;
   private SubscriberStub mockSubscriberStub;
+  private BidiStreamingCallable<StreamingPullRequest, StreamingPullResponse> mockStreamingCallable;
+  private UnaryCallable<AcknowledgeRequest, Empty> mockAcknowledgeCallable;
+  private UnaryCallable<ModifyAckDeadlineRequest, Empty> mockModifyAckDeadlineCallable;
 
   private static final String MOCK_SUBSCRIPTION_NAME =
       "projects/MOCK-PROJECT/subscriptions/MOCK-SUBSCRIPTION";
@@ -99,14 +104,28 @@ public class StreamingSubscriberConnectionTest {
   private static final Duration MAX_ACK_EXTENSION_PERIOD = Duration.ofMinutes(60);
 
   @Before
-  public void setUp() {
+  public void setUp() throws Exception {
     systemExecutor = new FakeScheduledExecutorService();
     executor = new FakeScheduledExecutorService();
     clock = systemExecutor.getClock();
-    mockSubscriberStub =
-        mock(
-            SubscriberStub.class,
-            withSettings().withoutAnnotations().defaultAnswer(RETURNS_DEEP_STUBS));
+
+    mockStreamingCallable = mock(BidiStreamingCallable.class, withSettings().withoutAnnotations());
+    mockAcknowledgeCallable = mock(UnaryCallable.class, withSettings().withoutAnnotations());
+    mockModifyAckDeadlineCallable = mock(UnaryCallable.class, withSettings().withoutAnnotations());
+
+    mockSubscriberStub = mock(SubscriberStub.class, withSettings().withoutAnnotations());
+    when(mockSubscriberStub.streamingPullCallable()).thenReturn(mockStreamingCallable);
+    when(mockSubscriberStub.acknowledgeCallable()).thenReturn(mockAcknowledgeCallable);
+    when(mockSubscriberStub.modifyAckDeadlineCallable()).thenReturn(mockModifyAckDeadlineCallable);
+
+    ClientStream<StreamingPullRequest> defaultMockClientStream =
+        mock(ClientStream.class, withSettings().withoutAnnotations());
+    when(mockStreamingCallable.splitCall(any(), any())).thenReturn(defaultMockClientStream);
+
+    when(mockAcknowledgeCallable.futureCall(any()))
+        .thenAnswer(invocation -> SettableApiFuture.create());
+    when(mockModifyAckDeadlineCallable.futureCall(any()))
+        .thenAnswer(invocation -> SettableApiFuture.create());
   }
 
   @After
@@ -391,12 +410,9 @@ public void testSendAckOperationsExactlyOnceEnabledMessageFuturesModacks() {
     systemExecutor.advanceTime(Duration.ofSeconds(200));
 
     // Assert expected behavior
-    verify(mockSubscriberStub.modifyAckDeadlineCallable(), times(1))
-        .futureCall(modifyAckDeadlineRequestNack);
-    verify(mockSubscriberStub.modifyAckDeadlineCallable(), times(1))
-        .futureCall(modifyAckDeadlineRequestInitial);
-    verify(mockSubscriberStub.modifyAckDeadlineCallable(), times(1))
-        .futureCall(modifyAckDeadlineRequestRetry);
+    verify(mockModifyAckDeadlineCallable, times(1)).futureCall(modifyAckDeadlineRequestNack);
+    verify(mockModifyAckDeadlineCallable, times(1)).futureCall(modifyAckDeadlineRequestInitial);
+    verify(mockModifyAckDeadlineCallable, times(1)).futureCall(modifyAckDeadlineRequestRetry);
     verify(mockSubscriberStub, never()).acknowledgeCallable();
 
     try {
@@ -511,9 +527,8 @@ public void testSendAckOperationsExactlyOnceEnabledMessageFuturesAcks() {
     systemExecutor.advanceTime(Duration.ofMillis(200));
 
     // Assert expected behavior;
-    verify(mockSubscriberStub.acknowledgeCallable(), times(1))
-        .futureCall(acknowledgeRequestInitial);
-    verify(mockSubscriberStub.acknowledgeCallable(), times(1))
+    verify(mockAcknowledgeCallable, times(1)).futureCall(acknowledgeRequestInitial);
+    verify(mockAcknowledgeCallable, times(1))
         .futureCall(
             argThat(new CustomArgumentMatchers.AcknowledgeRequestMatcher(acknowledgeRequestRetry)));
     verify(mockSubscriberStub, never()).modifyAckDeadlineCallable();
@@ -577,7 +592,7 @@ public void testSendAckOperationsExactlyOnceEnabledErrorWithEmptyMetadataMap() {
     systemExecutor.advanceTime(Duration.ofMillis(200));
 
     // Assert expected behavior;
-    verify(mockSubscriberStub.acknowledgeCallable(), times(2)).futureCall(acknowledgeRequest);
+    verify(mockAcknowledgeCallable, times(2)).futureCall(acknowledgeRequest);
     verify(mockSubscriberStub, never()).modifyAckDeadlineCallable();
 
     try {
@@ -677,27 +692,22 @@ public void testMaxPerRequestChanges() {
               .setSubscription(MOCK_SUBSCRIPTION_NAME)
               .addAllAckIds(mockAckIdsInRequest)
               .build();
-      verify(mockSubscriberStub.acknowledgeCallable(), times(1))
-          .futureCall(expectedAcknowledgeRequest);
+      verify(mockAcknowledgeCallable, times(1)).futureCall(expectedAcknowledgeRequest);
 
       ModifyAckDeadlineRequest expectedModifyAckDeadlineRequest =
           ModifyAckDeadlineRequest.newBuilder()
               .setSubscription(MOCK_SUBSCRIPTION_NAME)
               .addAllAckIds(mockAckIdsInRequest)
               .setAckDeadlineSeconds(MOCK_ACK_EXTENSION_DEFAULT_SECONDS)
               .build();
-      verify(mockSubscriberStub.modifyAckDeadlineCallable(), times(1))
-          .futureCall(expectedModifyAckDeadlineRequest);
+      verify(mockModifyAckDeadlineCallable, times(1)).futureCall(expectedModifyAckDeadlineRequest);
     }
   }
 
   @Test
   public void testClientPinger_pingSent() {
-    BidiStreamingCallable<StreamingPullRequest, StreamingPullResponse> mockStreamingCallable =
-        mock(BidiStreamingCallable.class, withSettings().withoutAnnotations());
     ClientStream<StreamingPullRequest> mockClientStream =
         mock(ClientStream.class, withSettings().withoutAnnotations());
-    when(mockSubscriberStub.streamingPullCallable()).thenReturn(mockStreamingCallable);
     when(mockStreamingCallable.splitCall(any(ResponseObserver.class), any()))
         .thenReturn(mockClientStream);
 
@@ -737,11 +747,8 @@ public void testClientPinger_pingSent() {
 
   @Test
   public void testClientPinger_pingsNotSentWhenDisabled() {
-    BidiStreamingCallable<StreamingPullRequest, StreamingPullResponse> mockStreamingCallable =
-        mock(BidiStreamingCallable.class, withSettings().withoutAnnotations());
     ClientStream<StreamingPullRequest> mockClientStream =
         mock(ClientStream.class, withSettings().withoutAnnotations());
-    when(mockSubscriberStub.streamingPullCallable()).thenReturn(mockStreamingCallable);
     when(mockStreamingCallable.splitCall(any(ResponseObserver.class), any()))
         .thenReturn(mockClientStream);
 
@@ -763,13 +770,10 @@ public void testClientPinger_pingsNotSentWhenDisabled() {
 
   @Test
   public void testServerMonitor_timesOut() {
-    BidiStreamingCallable<StreamingPullRequest, StreamingPullResponse> mockStreamingCallable =
-        mock(BidiStreamingCallable.class, withSettings().withoutAnnotations());
     ClientStream<StreamingPullRequest> mockClientStream =
         mock(ClientStream.class, withSettings().withoutAnnotations());
     ArgumentCaptor<ResponseObserver<StreamingPullResponse>> observerCaptor =
         ArgumentCaptor.forClass(ResponseObserver.class);
-    when(mockSubscriberStub.streamingPullCallable()).thenReturn(mockStreamingCallable);
     when(mockStreamingCallable.splitCall(observerCaptor.capture(), any()))
         .thenReturn(mockClientStream);
 
@@ -813,13 +817,10 @@ public void testServerMonitor_timesOut() {
 
   @Test
   public void testServerMonitor_doesNotTimeOutIfResponseReceived() {
-    BidiStreamingCallable<StreamingPullRequest, StreamingPullResponse> mockStreamingCallable =
-        mock(BidiStreamingCallable.class, withSettings().withoutAnnotations());
     ClientStream<StreamingPullRequest> mockClientStream =
         mock(ClientStream.class, withSettings().withoutAnnotations());
     ArgumentCaptor<ResponseObserver<StreamingPullResponse>> observerCaptor =
         ArgumentCaptor.forClass(ResponseObserver.class);
-    when(mockSubscriberStub.streamingPullCallable()).thenReturn(mockStreamingCallable);
     when(mockStreamingCallable.splitCall(observerCaptor.capture(), any()))
         .thenReturn(mockClientStream);