GH-637: Fix spurious nonresponsive consumer events

Fixes #637

Wrong timestamp used for event publication causing invalid events.

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -357,6 +357,8 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private boolean taskSchedulerExplicitlySet;
 
+		private volatile long lastPoll = System.currentTimeMillis();
+
 		@SuppressWarnings("unchecked")
 		ListenerConsumer(GenericMessageListener<?> listener, ListenerType listenerType) {
 			Assert.state(!this.isAnyManualAck || !this.autoCommit,
@@ -437,7 +439,7 @@ else if (listener instanceof MessageListener) {
 		}
 
 		protected void checkConsumer() {
-			long timeSinceLastPoll = System.currentTimeMillis() - last;
+			long timeSinceLastPoll = System.currentTimeMillis() - this.lastPoll;
 			if (((float) timeSinceLastPoll) / (float) this.containerProperties.getPollTimeout()
 					> this.containerProperties.getNoPollThreshold()) {
 				publishNonResponsiveConsumerEvent(timeSinceLastPoll, this.consumer);
@@ -619,6 +621,8 @@ public void run() {
 					}
 					processSeeks();
 					ConsumerRecords<K, V> records = this.consumer.poll(this.containerProperties.getPollTimeout());
+					this.lastPoll = System.currentTimeMillis();
+
 					if (records != null && this.logger.isDebugEnabled()) {
 						this.logger.debug("Received: " + records.count() + " records");
 					}

spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

@@ -647,6 +647,40 @@ public void testBrokerDownEvent() throws Exception {
 		container.stop();
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	public void testNonResponsiveConsumerEventNotIssuedWithActiveConsumer() throws Exception {
+		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
+		Consumer<Integer, String> consumer = mock(Consumer.class);
+		given(cf.createConsumer(isNull(), eq(""))).willReturn(consumer);
+		ConsumerRecords records = new ConsumerRecords(Collections.emptyMap());
+		CountDownLatch latch = new CountDownLatch(20);
+		given(consumer.poll(anyLong())).willAnswer(i -> {
+			Thread.sleep(100);
+			latch.countDown();
+			return records;
+		});
+		TopicPartitionInitialOffset[] topicPartition = new TopicPartitionInitialOffset[] {
+				new TopicPartitionInitialOffset("foo", 0) };
+		ContainerProperties containerProps = new ContainerProperties(topicPartition);
+		containerProps.setNoPollThreshold(2.0f);
+		containerProps.setPollTimeout(100);
+		containerProps.setMonitorInterval(1);
+		containerProps.setMessageListener(mock(MessageListener.class));
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		final AtomicInteger eventCounter = new AtomicInteger();
+		container.setApplicationEventPublisher(e -> {
+			if (e instanceof NonResponsiveConsumerEvent) {
+				eventCounter.incrementAndGet();
+			}
+		});
+		container.start();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		container.stop();
+		assertThat(eventCounter.get()).isEqualTo(0);
+	}
+
 	@Test
 	public void testBatchAck() throws Exception {
 		logger.info("Start batch ack");