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

* Fixed `KafkaMessageListenerContainerTests` for Java 7 and
appropriate Mockito version

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

@@ -359,6 +359,8 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private boolean taskSchedulerExplicitlySet;
 
+		private volatile long lastPoll = System.currentTimeMillis();
+
 		@SuppressWarnings("unchecked")
 		ListenerConsumer(GenericMessageListener<?> listener, GenericAcknowledgingMessageListener<?> ackListener) {
 			Assert.state(!this.isAnyManualAck || !this.autoCommit,
@@ -461,7 +463,7 @@ public void run() {
 		}
 
 		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);
@@ -624,6 +626,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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -430,6 +430,50 @@ public void publishEvent(ApplicationEvent event) {
 		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(anyString(), 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(new ApplicationEventPublisher() {
+
+			@Override
+			public void publishEvent(Object e) {
+				if (e instanceof NonResponsiveConsumerEvent) {
+					eventCounter.incrementAndGet();
+				}
+			}
+
+			@Override
+			public void publishEvent(ApplicationEvent event) {
+				publishEvent((Object) event);
+			}
+
+		});
+		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");