GH-112: Commit offsets even when fetch is paused

Fixes GH-112 (https://github.com/spring-projects/spring-kafka/issues/112)

Increase timeout on unrelated failing tests

* Polishing according PR comments

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

@@ -72,6 +72,7 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
  * @author Gary Russell
  * @author Murali Reddy
  * @author Marius Bogoevici
+ * @author Martin Dam
  */
 public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListenerContainer<K, V> {
 
@@ -428,7 +429,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 						}
 					}
 					this.unsent = checkPause(this.unsent);
-					if (!this.paused && !this.autoCommit) {
+					if (!this.autoCommit) {
 						processCommits();
 					}
 				}

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -113,7 +113,7 @@ public class EnableKafkaIntegrationTests {
 	public void testSimple() throws Exception {
 		template.send("annotated1", 0, "foo");
 		template.flush();
-		assertThat(this.listener.latch1.await(20, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.listener.latch1.await(60, TimeUnit.SECONDS)).isTrue();
 
 		template.send("annotated2", 0, 123, "foo");
 		template.flush();
@@ -182,21 +182,21 @@ public class EnableKafkaIntegrationTests {
 	public void testInterface() throws Exception {
 		template.send("annotated7", 0, "foo");
 		template.flush();
-		assertThat(this.ifaceListener.getLatch1().await(20, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.ifaceListener.getLatch1().await(60, TimeUnit.SECONDS)).isTrue();
 	}
 
 	@Test
 	public void testMulti() throws Exception {
 		template.send("annotated8", 0, "foo");
 		template.flush();
-		assertThat(this.multiListener.latch1.await(20, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.multiListener.latch1.await(60, TimeUnit.SECONDS)).isTrue();
 	}
 
 	@Test
 	public void testTx() throws Exception {
 		template.send("annotated9", 0, "foo");
 		template.flush();
-		assertThat(this.ifaceListener.getLatch2().await(20, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.ifaceListener.getLatch2().await(60, TimeUnit.SECONDS)).isTrue();
 	}
 
 	@Test

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

@@ -59,7 +59,7 @@ import org.springframework.retry.support.RetryTemplate;
  * Tests for the listener container.
  *
  * @author Gary Russell
- *
+ * @author Martin Dam
  */
 public class KafkaMessageListenerContainerTests {
 
@@ -73,8 +73,10 @@ public class KafkaMessageListenerContainerTests {
 
 	private static String topic4 = "testTopic4";
 
+	private static String topic5 = "testTopic5";
+
 	@ClassRule
-	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4);
+	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5);
 
 	@Rule
 	public TestName testName = new TestName();
@@ -192,6 +194,54 @@ public class KafkaMessageListenerContainerTests {
 		logger.info("Stop " + this.testName.getMethodName() + ackMode);
 	}
 
+	@Test
+	public void testSlowConsumerCommitsAreProcessed() throws Exception {
+		Map<String, Object> props = KafkaTestUtils.consumerProps("slow", "false", embeddedKafka);
+		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
+		ContainerProperties containerProps = new ContainerProperties(topic5);
+		containerProps.setAckCount(1);
+		containerProps.setPauseAfter(100);
+		containerProps.setAckMode(AckMode.MANUAL);
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		final CountDownLatch latch = new CountDownLatch(3);
+		containerProps.setMessageListener((AcknowledgingMessageListener<Integer, String>) (message, ack) -> {
+			logger.info("slow: " + message);
+			try {
+				Thread.sleep(1000);
+			}
+			catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+			}
+			ack.acknowledge();
+			latch.countDown();
+		});
+		container.setBeanName("testSlow");
+		container.start();
+		ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
+		Consumer<?, ?> consumer = spyOnConsumer(container);
+
+
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
+		template.setDefaultTopic(topic5);
+		template.sendDefault(0, "foo");
+		template.sendDefault(2, "bar");
+		template.flush();
+		Thread.sleep(300);
+		template.sendDefault(0, "fiz");
+		template.sendDefault(2, "buz");
+		template.flush();
+
+		// Verify that commitSync is called when paused
+		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
+		verify(consumer, atLeastOnce()).pause(any(TopicPartition.class), any(TopicPartition.class));
+		verify(consumer, atLeastOnce()).commitSync(any());
+		verify(consumer, atLeastOnce()).resume(any(TopicPartition.class), any(TopicPartition.class));
+		container.stop();
+	}
+
 	@Test
 	public void testSlowConsumerWithException() throws Exception {
 		logger.info("Start " + this.testName.getMethodName());