GH-1247: Polishing

- revert lastPoll position
- fix return/break issue
- fix race in existing test

Author: garyrussell

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

@@ -730,7 +730,6 @@ public void run() {
 						}
 						publishConsumerPausedEvent(this.consumer.assignment());
 					}
-					this.lastPoll = System.currentTimeMillis();
 					this.polling.set(true);
 					ConsumerRecords<K, V> records = this.consumer.poll(this.containerProperties.getPollTimeout());
 					if (!this.polling.compareAndSet(true, false)) {
@@ -741,8 +740,9 @@ public void run() {
 						if (records.count() > 0 && this.logger.isDebugEnabled()) {
 							this.logger.debug("Discarding polled records, container stopped: " + records.count());
 						}
-						return;
+						break;
 					}
+					this.lastPoll = System.currentTimeMillis();
 					if (this.consumerPaused && !isPaused()) {
 						if (this.logger.isDebugEnabled()) {
 							this.logger.debug("Resuming consumption from: " + this.consumer.paused());

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

@@ -551,7 +551,6 @@ public void onMessage(ConsumerRecord<Integer, String> data) {
 		inOrder.verify(messageListener).onMessage(any(ConsumerRecord.class));
 		inOrder.verify(consumer).commitSync(any(Map.class));
 		container.stop();
-		verify(consumer).wakeup();
 	}
 
 	@SuppressWarnings("unchecked")