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

	final String testTopic = "test-topic";
	// -------- old partitions --------
	KafkaTopicPartitionState<TopicPartition> oldPartition1 = new KafkaTopicPartitionState<>(
		new KafkaTopicPartition(testTopic, 0), new TopicPartition(testTopic, 0));
	oldPartition1.setOffset(23L);
	KafkaTopicPartitionState<TopicPartition> oldPartition2 = new KafkaTopicPartitionState<>(
		new KafkaTopicPartition(testTopic, 1), new TopicPartition(testTopic, 1));
	oldPartition2.setOffset(32L);
	List<KafkaTopicPartitionState<TopicPartition>> oldPartitions = new ArrayList<>(2);
	oldPartitions.add(oldPartition1);
	oldPartitions.add(oldPartition2);
	// -------- new partitions with defined offsets --------
	KafkaTopicPartitionState<TopicPartition> newPartition = new KafkaTopicPartitionState<>(
		new KafkaTopicPartition(testTopic, 2), new TopicPartition(testTopic, 2));
	newPartition.setOffset(29L);
	// -------- setup mock KafkaConsumer --------
	// initial assignments
	final Map<TopicPartition, Long> mockConsumerAssignmentsToPositions = new LinkedHashMap<>();
	for (KafkaTopicPartitionState<TopicPartition> oldPartition : oldPartitions) {
		mockConsumerAssignmentsToPositions.put(oldPartition.getKafkaPartitionHandle(), oldPartition.getOffset() + 1);
	}
	final KafkaConsumer<byte[], byte[]> mockConsumer = createMockConsumer(
			mockConsumerAssignmentsToPositions,
			Collections.<TopicPartition, Long>emptyMap(),
			true,
			null,
			null);
	// -------- setup new partitions to be polled from the unassigned partitions queue --------
	final ClosableBlockingQueue<KafkaTopicPartitionState<TopicPartition>> unassignedPartitionsQueue =
		new ClosableBlockingQueue<>();
	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));
	verify(mockConsumer, times(1)).wakeup();
	testThread.startPartitionReassignment();
	testThread.waitPartitionReassignmentComplete();
	// the consumer's assignment should have remained untouched
	assertEquals(oldPartitions.size(), mockConsumerAssignmentsToPositions.size());
	for (KafkaTopicPartitionState<TopicPartition> oldPartition : oldPartitions) {
		assertTrue(mockConsumerAssignmentsToPositions.containsKey(oldPartition.getKafkaPartitionHandle()));
		assertEquals(
				oldPartition.getOffset() + 1,
				mockConsumerAssignmentsToPositions.get(oldPartition.getKafkaPartitionHandle()).longValue());
	}
	// the new partitions should have been re-added to the unassigned partitions queue
	assertEquals(1, unassignedPartitionsQueue.size());
}