@SuppressWarnings("unchecked")
@Test(timeout = 10000)
public void testReassignPartitionsDefinedOffsetsWithoutInitialAssignmentsWhenEarlyWakeup() throws Exception {

	final String testTopic = "test-topic";
	// -------- new partitions with defined offsets --------
	KafkaTopicPartitionState<TopicPartition> newPartition1 = new KafkaTopicPartitionState<>(
		new KafkaTopicPartition(testTopic, 0), new TopicPartition(testTopic, 0));
	newPartition1.setOffset(KafkaTopicPartitionStateSentinel.EARLIEST_OFFSET);
	KafkaTopicPartitionState<TopicPartition> newPartition2 = new KafkaTopicPartitionState<>(
		new KafkaTopicPartition(testTopic, 1), new TopicPartition(testTopic, 1));
	newPartition2.setOffset(KafkaTopicPartitionStateSentinel.EARLIEST_OFFSET);
	List<KafkaTopicPartitionState<TopicPartition>> newPartitions = new ArrayList<>(2);
	newPartitions.add(newPartition1);
	newPartitions.add(newPartition2);
	// -------- setup mock KafkaConsumer --------
	// no initial assignments
	final Map<TopicPartition, Long> mockConsumerAssignmentsAndPositions = new LinkedHashMap<>();
	// mock retrieved values that should replace the EARLIEST_OFFSET sentinels
	final Map<TopicPartition, Long> mockRetrievedPositions = new HashMap<>();
	mockRetrievedPositions.put(newPartition1.getKafkaPartitionHandle(), 23L);
	mockRetrievedPositions.put(newPartition2.getKafkaPartitionHandle(), 32L);
	final KafkaConsumer<byte[], byte[]> mockConsumer = createMockConsumer(
			mockConsumerAssignmentsAndPositions,
			mockRetrievedPositions,
			true,
			null,
			null);
	// -------- setup new partitions to be polled from the unassigned partitions queue --------
	final ClosableBlockingQueue<KafkaTopicPartitionState<TopicPartition>> unassignedPartitionsQueue =
		new ClosableBlockingQueue<>();
	for (KafkaTopicPartitionState<TopicPartition> newPartition : newPartitions) {
		unassignedPartitionsQueue.add(newPartition);
	}
	// -------- start test --------
	final TestKafkaConsumerThread testThread =
		new TestKafkaConsumerThread(mockConsumer, unassignedPartitionsQueue, new Handover());
	testThread.start();
	// pause just before the reassignment so we can inject the wakeup
	testThread.waitPartitionReassignmentInvoked();
	testThread.setOffsetsToCommit(new HashMap<TopicPartition, OffsetAndMetadata>(), mock(KafkaCommitCallback.class));
	// make sure the consumer was actually woken up
	verify(mockConsumer, times(1)).wakeup();
	testThread.startPartitionReassignment();
	testThread.waitPartitionReassignmentComplete();
	// the consumer's assignment should have remained untouched (in this case, empty)
	assertEquals(0, mockConsumerAssignmentsAndPositions.size());
	// the new partitions should have been re-added to the unassigned partitions queue
	assertEquals(2, unassignedPartitionsQueue.size());
}