@SuppressWarnings("unchecked")
@Test(timeout = 10000)
public void testReassignPartitionsDefinedOffsetsWithoutInitialAssignmentsWhenWakeupMidway() 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);
	// these latches are used to pause midway the reassignment process
	final OneShotLatch midAssignmentLatch = new OneShotLatch();
	final OneShotLatch continueAssigmentLatch = new OneShotLatch();
	final KafkaConsumer<byte[], byte[]> mockConsumer = createMockConsumer(
			mockConsumerAssignmentsAndPositions,
			mockRetrievedPositions,
			false,
			midAssignmentLatch,
			continueAssigmentLatch);
	// -------- 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();
	testThread.startPartitionReassignment();
	// wait until the reassignment has started
	midAssignmentLatch.await();
	testThread.setOffsetsToCommit(new HashMap<TopicPartition, OffsetAndMetadata>(), mock(KafkaCommitCallback.class));
	// the wakeup in the setOffsetsToCommit() call should have been buffered, and not called on the consumer
	verify(mockConsumer, never()).wakeup();
	continueAssigmentLatch.trigger();
	testThread.waitPartitionReassignmentComplete();
	// verify that the consumer called assign() with all new partitions, and that positions are correctly advanced
	assertEquals(newPartitions.size(), mockConsumerAssignmentsAndPositions.size());
	for (KafkaTopicPartitionState<TopicPartition> newPartition : newPartitions) {
		assertTrue(mockConsumerAssignmentsAndPositions.containsKey(newPartition.getKafkaPartitionHandle()));
		// should be seeked to (offset in state + 1) because offsets in state represent the last processed record
		assertEquals(
			newPartition.getOffset() + 1,
			mockConsumerAssignmentsAndPositions.get(newPartition.getKafkaPartitionHandle()).longValue());
	}
	// after the reassignment, the consumer should be restored the wakeup call
	verify(mockConsumer, times(1)).wakeup();
	assertEquals(0, unassignedPartitionsQueue.size());
}