diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
index 7a0a5cdf..454b8272 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
+++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
@@ -72,9 +72,11 @@ public class ConcurrentMessageListenerContainerTests {
 
 	private static String topic6 = "testTopic6";
 
+	private static String topic7 = "testTopic7";
+
 	@ClassRule
 	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5,
-			topic6);
+			topic6, topic7);
 
 	@Test
 	public void testAutoCommit() throws Exception {
@@ -278,7 +280,7 @@ public class ConcurrentMessageListenerContainerTests {
 
 	private void testManualCommitGuts(AckMode ackMode, String topic) throws Exception {
 		logger.info("Start " + ackMode);
-		Map<String, Object> props = KafkaTestUtils.consumerProps("test4", "false", embeddedKafka);
+		Map<String, Object> props = KafkaTestUtils.consumerProps("test" + ackMode, "false", embeddedKafka);
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props);
 		ConcurrentMessageListenerContainer<Integer, String> container =
 				new ConcurrentMessageListenerContainer<>(cf, topic);
@@ -312,6 +314,49 @@ public class ConcurrentMessageListenerContainerTests {
 		logger.info("Stop " + ackMode);
 	}
 
+	@Test
+	public void testManualCommitExisting() throws Exception {
+		logger.info("Start MANUAL_IMMEDIATE with Existing");
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
+		template.setDefaultTopic(topic7);
+		template.send(0, "foo");
+		template.send(2, "bar");
+		template.send(0, "baz");
+		template.send(2, "qux");
+		template.flush();
+		Map<String, Object> props = KafkaTestUtils.consumerProps("testManualExisting", "false", embeddedKafka);
+		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props);
+		ConcurrentMessageListenerContainer<Integer, String> container =
+				new ConcurrentMessageListenerContainer<>(cf, topic7);
+		final CountDownLatch latch = new CountDownLatch(8);
+		container.setMessageListener(new AcknowledgingMessageListener<Integer, String>() {
+
+			@Override
+			public void onMessage(ConsumerRecord<Integer, String> message, Acknowledgment ack) {
+				logger.info("manualExisting: " + message);
+				ack.acknowledge();
+				latch.countDown();
+			}
+
+		});
+		container.setConcurrency(1);
+		container.setAckMode(AckMode.MANUAL_IMMEDIATE);
+		container.setBeanName("testManualExisting");
+		container.start();
+		ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
+		template.send(0, "fooo");
+		template.send(2, "barr");
+		template.send(0, "bazz");
+		template.send(2, "quxx");
+		template.flush();
+		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
+		container.stop();
+		logger.info("Stop MANUAL_IMMEDIATE with Existing");
+	}
+
 
 	@SuppressWarnings("unchecked")
 	@Test