159160161162163164165166167168169
rateLimiter.setRate(Double.MAX_VALUE); run.set(false); t.join(); consumer.shutdown(); sink.close(); assertEquals(sink.getNumOfPendingMessages(), 0); }