@Test(timeout = 10000)
public void testCloseWithoutAssignedPartitions() throws Exception {

	// no initial assignment
	final KafkaConsumer<byte[], byte[]> mockConsumer = createMockConsumer(
		new LinkedHashMap<TopicPartition, Long>(),
		Collections.<TopicPartition, Long>emptyMap(),
		false,
		null,
		null);
	// setup latch so the test waits until testThread is blocked on getBatchBlocking method
	final MultiShotLatch getBatchBlockingInvoked = new MultiShotLatch();
	final ClosableBlockingQueue<KafkaTopicPartitionState<TopicPartition>> unassignedPartitionsQueue =
		new ClosableBlockingQueue<KafkaTopicPartitionState<TopicPartition>>() {
			@Override
			public List<KafkaTopicPartitionState<TopicPartition>> getBatchBlocking() throws InterruptedException {
				getBatchBlockingInvoked.trigger();
				return super.getBatchBlocking();
			}
		};
	final TestKafkaConsumerThread testThread =
		new TestKafkaConsumerThread(mockConsumer, unassignedPartitionsQueue, new Handover());
	testThread.start();
	getBatchBlockingInvoked.await();
	testThread.shutdown();
	testThread.join();
}