code: public CompletableFuture<Boolean> drain(Duration timeout) throws InterruptedException {
         if (!this.isActive() || this.connection==null) {
             throw new IllegalStateException("Consumer is closed");
         }
  
         if (isDraining()) {
             return this.getDrainingFuture();
         }
  
         Instant start = Instant.now();
         final CompletableFuture<Boolean> tracker = new CompletableFuture<>();
         this.markDraining(tracker);
         this.sendUnsubForDrain();
  
         try {
              this.connection.flush(timeout); // Flush and wait up to the timeout
         } catch (TimeoutException e) {
             this.connection.processException(e);
         }
  
         this.markUnsubedForDrain();
  
          // Wait for the timeout or the pending count to go to 0, skipped if conn is
          // draining
          connection.getExecutor().submit(() -> {
              try {
                  Instant now = Instant.now();
  
                  while (timeout == null || timeout.equals(Duration.ZERO)
                          || Duration.between(start, now).compareTo(timeout) < 0) {
                      if (this.isDrained()) {
                          break;
                      }
  
                      Thread.sleep(1); // Sleep 1 milli
  
                      now = Instant.now();
                  }
  
                  this.cleanUpAfterDrain();
              } catch (InterruptedException e) {
                  this.connection.processException(e);
              } finally {
                  tracker.complete(this.isDrained());
              }
         });
  
         return getDrainingFuture();
     }
docs: Drain tells the consumer to process in flight, or cached messages, but stop receiving new ones. The library will
  flush the unsubscribe call(s) insuring that any publish calls made by this client are included. When all messages
  are processed the consumer effectively becomes unsubscribed.
  
  @param timeout The time to wait for the drain to succeed, pass 0 to wait
  forever. Drain involves moving messages to and from the server
  so a very short timeout is not recommended.
  @return A future that can be used to check if the drain has completed
  @throws InterruptedException if the thread is interrupted