### Question:
KafkaConsumerThread extends Thread { public void shutdown() { running = false; unassignedPartitionsQueue.close(); handover.wakeupProducer(); synchronized (consumerReassignmentLock) { if (consumer != null) { consumer.wakeup(); } else { hasBufferedWakeup = true; } } } KafkaConsumerThread(
			Logger log,
			Handover handover,
			Properties kafkaProperties,
			ClosableBlockingQueue<KafkaTopicPartitionState<TopicPartition>> unassignedPartitionsQueue,
			KafkaConsumerCallBridge consumerCallBridge,
			String threadName,
			long pollTimeout,
			boolean useMetrics,
			MetricGroup consumerMetricGroup,
			MetricGroup subtaskMetricGroup); @Override void run(); void shutdown();  }### Answer:
@Test(timeout = 10000) public void testCloseWithoutAssignedPartitions() throws Exception { final KafkaConsumer<byte[], byte[]> mockConsumer = createMockConsumer( new LinkedHashMap<TopicPartition, Long>(), Collections.<TopicPartition, Long>emptyMap(), false, null, null); 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(); }