SendToKafka extends TimerTask { @Override public void run() { long numSentCurrent = numSent.get(); long numSentSince = numSentCurrent - numSentLast; boolean sendMessages = numSentLast == 0 || numSentSince >= numMessagesSent; if(sendMessages) { Collection<Future<Long>> futures = Collections.synchronizedList(new ArrayList<>()); for(int batch = 0;batch < numBatches;++batch) { try { futures.add(pool.submit(() -> { KafkaProducer<String, String> producer = kafkaProducer.get(); Collection<Future<?>> b = Collections.synchronizedCollection(new ArrayList<>()); for (int i = 0; i < batchSize; ++i) { b.add(sendToKafka(producer, kafkaTopic, messageSupplier.get())); } for(Future<?> f : b) { f.get(); } return batchSize; })); } catch (Exception e) { e.printStackTrace(System.err); } } for(Future<Long> f : futures) { try { f.get(); } catch (Exception e) { e.printStackTrace(System.err); } } numSentLast = numSentCurrent; } } SendToKafka( String kafkaTopic
                    , long numMessagesSent
                    , int numBatches
                    , Supplier<String> messageSupplier
                    , ExecutorService pool
                    , AtomicLong numSent
                    , ThreadLocal<KafkaProducer<String, String>> kafkaProducer
                    ); @Override void run();  }
@Test public void testWritesCorrectNumber() { ExecutorService executor = ForkJoinPool.commonPool(); AtomicLong numSent = new AtomicLong(0); long expectedSent = 100; SendToKafka sender = new SendToKafka(null, expectedSent, 10, () -> "msg", executor, numSent, ThreadLocal.withInitial(() -> null) ) { @Override protected Future<?> sendToKafka(KafkaProducer producer, String kafkaTopic, String message) { assertEquals(message, "msg"); return ForkJoinPool.commonPool().submit(() -> { numSent.incrementAndGet(); }); } }; sender.run(); assertEquals(numSent.get(), expectedSent); }