ConsumerDriver extends Driver { public long getMessagesReceived() { return recordsFetched; } ConsumerDriver(ConsumerDefinition consumerDefinition);  ConsumerDriver(ConsumerDefinition consumerDefinition, CountDownLatch completionLatch); void setConsumerRebalanceListener(ConsumerRebalanceListener consumerRebalanceListener); @Override void drive(); long getMessagesReceived(); long getMessagesExpected();  }
@Test public void testReceive_sequential() throws InterruptedException { String topic = "foo"; int messageCount = 1000; CountDownLatch latch = new CountDownLatch(2); ProducerDriver producerDriver = createProducerDriver(latch, topic, messageCount); producerDriver.run(); ConsumerDriver consumerDriver = createConsumerDriver(latch, topic, messageCount); consumerDriver.run(); assertEquals(messageCount, consumerDriver.getMessagesReceived()); }