public void sendAcks(List<String> messagesToAck) throws IOException {
    try (SubscriberStub subscriber = pubsubQueueClient.getSubscriber(subscriberStubSettings)) {
      int numberOfBatches = (int) Math.ceil((double) messagesToAck.size() / DEFAULT_BATCH_SIZE_ACK_API);
      CompletableFuture.allOf(IntStream.range(0, numberOfBatches)
              .parallel()
              .boxed()
              .map(batchIndex -> getTask(subscriber, messagesToAck, batchIndex)).toArray(CompletableFuture[]::new))
          .get(MAX_WAIT_TIME_TO_ACK_MESSAGES, TimeUnit.MILLISECONDS);
      LOG.debug("Flushed out all outstanding acknowledged messages: " + messagesToAck.size());
    } catch (ExecutionException | InterruptedException | TimeoutException e) {
      throw new IOException("Failed to ack messages from PubSub", e);
    }
  }
@Test
  public void testSendAcks() throws IOException {
    doNothing().when(mockSubscriber).close();
    when(mockPubsubQueueClient.getSubscriber(any())).thenReturn(mockSubscriber);
    List<String> messageAcks = IntStream.range(0, 20).mapToObj(i -> "msg_" + i).collect(Collectors.toList());
    doNothing().when(mockPubsubQueueClient).makeAckRequest(eq(mockSubscriber), eq(SUBSCRIPTION_NAME), any());
    PubsubMessagesFetcher fetcher = new PubsubMessagesFetcher(
        PROJECT_ID, SUBSCRIPTION_ID, SMALL_BATCH_SIZE,
        MAX_MESSAGES_IN_REQUEST, MAX_WAIT_TIME_IN_REQUEST, mockPubsubQueueClient
    );

    fetcher.sendAcks(messageAcks);
    verify(mockPubsubQueueClient, times(2)).makeAckRequest(eq(mockSubscriber), eq(SUBSCRIPTION_NAME), any());
  }