public List<ReceivedMessage> fetchMessages() {
    List<ReceivedMessage> messageList = new ArrayList<>();
    try (SubscriberStub subscriber = pubsubQueueClient.getSubscriber(subscriberStubSettings)) {
      String subscriptionName = ProjectSubscriptionName.format(googleProjectId, pubsubSubscriptionId);
      long startTime = System.currentTimeMillis();
      long unAckedMessages = pubsubQueueClient.getNumUnAckedMessages(this.pubsubSubscriptionId);
      LOG.info("Found unacked messages " + unAckedMessages);
      while (messageList.size() < unAckedMessages && messageList.size() < maxMessagesPerSync
          && ((System.currentTimeMillis() - startTime) < (maxFetchTimePerSyncSecs * 1000))) {
        PullResponse pullResponse = pubsubQueueClient.makePullRequest(subscriber, subscriptionName, batchSize);
        messageList.addAll(pullResponse.getReceivedMessagesList());
      }
      return messageList;
    } catch (Exception e) {
      throw new HoodieException("Error when fetching metadata", e);
    }
  }
@Test
  public void testFetchMessages() throws IOException {
    doNothing().when(mockSubscriber).close();
    when(mockPubsubQueueClient.getSubscriber(any())).thenReturn(mockSubscriber);
    when(mockPubsubQueueClient.getNumUnAckedMessages(SUBSCRIPTION_ID)).thenReturn(3L);
    doNothing().when(mockSubscriber).close();
    ReceivedMessage message1 = ReceivedMessage.newBuilder().setAckId("1").setMessage(PubsubMessage.newBuilder().setMessageId("msgId1").build()).build();
    ReceivedMessage message2 = ReceivedMessage.newBuilder().setAckId("2").setMessage(PubsubMessage.newBuilder().setMessageId("msgId2").build()).build();
    ReceivedMessage message3 = ReceivedMessage.newBuilder().setAckId("3").setMessage(PubsubMessage.newBuilder().setMessageId("msgId3").build()).build();
    when(mockPubsubQueueClient.makePullRequest(mockSubscriber, SUBSCRIPTION_NAME, SMALL_BATCH_SIZE))
        .thenReturn(PullResponse.newBuilder().addReceivedMessages(message1).build())
        .thenReturn(PullResponse.newBuilder().addReceivedMessages(message2).build())
        .thenReturn(PullResponse.newBuilder().addReceivedMessages(message3).build());

    PubsubMessagesFetcher fetcher = new PubsubMessagesFetcher(
        PROJECT_ID, SUBSCRIPTION_ID, SMALL_BATCH_SIZE,
        MAX_MESSAGES_IN_REQUEST, MAX_WAIT_TIME_IN_REQUEST, mockPubsubQueueClient
    );
    List<ReceivedMessage> messages = fetcher.fetchMessages();

    assertEquals(3, messages.size());
    assertEquals("1", messages.get(0).getAckId());
    assertEquals("2", messages.get(1).getAckId());
    assertEquals("3", messages.get(2).getAckId());
    verify(mockPubsubQueueClient, times(3)).makePullRequest(mockSubscriber, SUBSCRIPTION_NAME, SMALL_BATCH_SIZE);
  }