KafkaConsumerThread extends Thread { void setOffsetsToCommit( Map<TopicPartition, OffsetAndMetadata> offsetsToCommit, @Nonnull KafkaCommitCallback commitCallback) { if (nextOffsetsToCommit.getAndSet(Tuple2.of(offsetsToCommit, commitCallback)) != null) { log.warn("Committing offsets to Kafka takes longer than the checkpoint interval. " + "Skipping commit of previous offsets because newer complete checkpoint offsets are available. " + "This does not compromise Flink's checkpoint integrity."); } 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 subtaskMetricGroup); @Override void run(); void shutdown();  }
@SuppressWarnings("unchecked") @Test(timeout = 10000) public void testReassigningPartitionsWithDefinedOffsetsWhenEarlyWakeup() throws Exception { final String testTopic = "test-topic"; 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); KafkaTopicPartitionState<TopicPartition> newPartition = new KafkaTopicPartitionState<>( new KafkaTopicPartition(testTopic, 2), new TopicPartition(testTopic, 2)); newPartition.setOffset(29L); 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); final ClosableBlockingQueue<KafkaTopicPartitionState<TopicPartition>> unassignedPartitionsQueue = new ClosableBlockingQueue<>(); unassignedPartitionsQueue.add(newPartition); final TestKafkaConsumerThread testThread = new TestKafkaConsumerThread(mockConsumer, unassignedPartitionsQueue, new Handover()); testThread.start(); testThread.waitPartitionReassignmentInvoked(); testThread.setOffsetsToCommit(new HashMap<TopicPartition, OffsetAndMetadata>(), mock(KafkaCommitCallback.class)); verify(mockConsumer, times(1)).wakeup(); testThread.startPartitionReassignment(); testThread.waitPartitionReassignmentComplete(); assertEquals(oldPartitions.size(), mockConsumerAssignmentsToPositions.size()); for (KafkaTopicPartitionState<TopicPartition> oldPartition : oldPartitions) { assertTrue(mockConsumerAssignmentsToPositions.containsKey(oldPartition.getKafkaPartitionHandle())); assertEquals( oldPartition.getOffset() + 1, mockConsumerAssignmentsToPositions.get(oldPartition.getKafkaPartitionHandle()).longValue()); } assertEquals(1, unassignedPartitionsQueue.size()); }
@SuppressWarnings("unchecked") @Test(timeout = 10000) public void testReassignPartitionsDefinedOffsetsWithoutInitialAssignmentsWhenEarlyWakeup() throws Exception { final String testTopic = "test-topic"; 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); final Map<TopicPartition, Long> mockConsumerAssignmentsAndPositions = new LinkedHashMap<>(); 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); final ClosableBlockingQueue<KafkaTopicPartitionState<TopicPartition>> unassignedPartitionsQueue = new ClosableBlockingQueue<>(); for (KafkaTopicPartitionState<TopicPartition> newPartition : newPartitions) { unassignedPartitionsQueue.add(newPartition); } final TestKafkaConsumerThread testThread = new TestKafkaConsumerThread(mockConsumer, unassignedPartitionsQueue, new Handover()); testThread.start(); testThread.waitPartitionReassignmentInvoked(); testThread.setOffsetsToCommit(new HashMap<TopicPartition, OffsetAndMetadata>(), mock(KafkaCommitCallback.class)); verify(mockConsumer, times(1)).wakeup(); testThread.startPartitionReassignment(); testThread.waitPartitionReassignmentComplete(); assertEquals(0, mockConsumerAssignmentsAndPositions.size()); assertEquals(2, unassignedPartitionsQueue.size()); }
@SuppressWarnings("unchecked") @Test(timeout = 10000) public void testReassignPartitionsDefinedOffsetsWithoutInitialAssignmentsWhenWakeupMidway() throws Exception { final String testTopic = "test-topic"; 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); final Map<TopicPartition, Long> mockConsumerAssignmentsAndPositions = new LinkedHashMap<>(); final Map<TopicPartition, Long> mockRetrievedPositions = new HashMap<>(); mockRetrievedPositions.put(newPartition1.getKafkaPartitionHandle(), 23L); mockRetrievedPositions.put(newPartition2.getKafkaPartitionHandle(), 32L); final OneShotLatch midAssignmentLatch = new OneShotLatch(); final OneShotLatch continueAssigmentLatch = new OneShotLatch(); final KafkaConsumer<byte[], byte[]> mockConsumer = createMockConsumer( mockConsumerAssignmentsAndPositions, mockRetrievedPositions, false, midAssignmentLatch, continueAssigmentLatch); final ClosableBlockingQueue<KafkaTopicPartitionState<TopicPartition>> unassignedPartitionsQueue = new ClosableBlockingQueue<>(); for (KafkaTopicPartitionState<TopicPartition> newPartition : newPartitions) { unassignedPartitionsQueue.add(newPartition); } final TestKafkaConsumerThread testThread = new TestKafkaConsumerThread(mockConsumer, unassignedPartitionsQueue, new Handover()); testThread.start(); testThread.startPartitionReassignment(); midAssignmentLatch.await(); testThread.setOffsetsToCommit(new HashMap<TopicPartition, OffsetAndMetadata>(), mock(KafkaCommitCallback.class)); verify(mockConsumer, never()).wakeup(); continueAssigmentLatch.trigger(); testThread.waitPartitionReassignmentComplete(); assertEquals(newPartitions.size(), mockConsumerAssignmentsAndPositions.size()); for (KafkaTopicPartitionState<TopicPartition> newPartition : newPartitions) { assertTrue(mockConsumerAssignmentsAndPositions.containsKey(newPartition.getKafkaPartitionHandle())); assertEquals( newPartition.getOffset() + 1, mockConsumerAssignmentsAndPositions.get(newPartition.getKafkaPartitionHandle()).longValue()); } verify(mockConsumer, times(1)).wakeup(); assertEquals(0, unassignedPartitionsQueue.size()); }