diff --git a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractTwitterMessageSource.java b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractTwitterMessageSource.java
index c49b093199..8f63ff534e 100644
--- a/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractTwitterMessageSource.java
+++ b/spring-integration-twitter/src/main/java/org/springframework/integration/twitter/inbound/AbstractTwitterMessageSource.java
@@ -64,6 +64,8 @@ public abstract class AbstractTwitterMessageSource<T> extends AbstractEndpoint
 	protected volatile int prefetchThreshold = 0;
 
 	protected volatile long markerId = -1;
+	
+	protected volatile long processedId = -1;
 
 	protected final TwitterOperations twitter;
 
@@ -124,7 +126,7 @@ public abstract class AbstractTwitterMessageSource<T> extends AbstractEndpoint
 		String lastId = this.metadataStore.get(this.metadataKey);
 		// initialize the last status ID from the metadataStore
 		if (StringUtils.hasText(lastId)){
-			this.markLastStatusId(Long.parseLong(lastId));
+			this.markerId = Long.parseLong(lastId);
 		}
 	}
 
@@ -166,7 +168,7 @@ public abstract class AbstractTwitterMessageSource<T> extends AbstractEndpoint
 	public Message<?> receive() {
 		Tweet tweet = tweets.poll();	
 		if (tweet != null){
-			
+			this.markProcessedId(tweet.getId());
 			return MessageBuilder.withPayload(tweet).build();
 		}
 		return null;
@@ -176,22 +178,16 @@ public abstract class AbstractTwitterMessageSource<T> extends AbstractEndpoint
 		synchronized (this.markerGuard) {
 			
 			long id = tweet.getId();
-			
-			String lastId = this.metadataStore.get(this.metadataKey);
 
-			long lastTweetId = 0;
-			if (lastId != null) {
-				lastTweetId = Long.parseLong(lastId);
-			}
-			if (id > lastTweetId) {
-				markLastStatusId(tweet.getId());
+			if (id > this.markerId) {
+				this.markerId = id;
 				tweets.add(tweet);
 			}
 		}
 	}
 
-	protected void markLastStatusId(long statusId) {
-		this.markerId = statusId;
+	protected void markProcessedId(long statusId) {
+		this.processedId = statusId;
 		this.metadataStore.put(this.metadataKey, String.valueOf(statusId));
 	}
 }
diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java
index c101509578..c48a4da6c8 100644
--- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java
+++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/DirectMessageReceivingMessageSourceTests.java
@@ -33,7 +33,9 @@ import java.util.Queue;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
+
 import org.springframework.beans.factory.support.DefaultListableBeanFactory;
+import org.springframework.integration.Message;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.store.PropertiesPersistingMetadataStore;
 import org.springframework.integration.test.util.TestUtils;
@@ -131,8 +133,8 @@ public class DirectMessageReceivingMessageSourceTests {
 		Tweet message = (Tweet) msg.poll();
 		assertEquals(2000, message.getId());
 		Thread.sleep(1000);
-		verify(twitter, times(1)).getDirectMessages(2000);
-		// based on the Mock, the Queue shoud now have 2 mopre messages third and fourth
+		verify(twitter, times(1)).getDirectMessages();
+		// based on the Mock, the Queue shoud now have 1 more messages third and fourth
 		assertTrue(((Queue)TestUtils.getPropertyValue(source, "tweets")).size() == 2);
 		source.stop();
 	}
@@ -165,8 +167,9 @@ public class DirectMessageReceivingMessageSourceTests {
 		Queue msg = (Queue) TestUtils.getPropertyValue(source, "tweets");
 		assertTrue(!CollectionUtils.isEmpty(msg));	
 		assertEquals(1, msg.size()); 
-		Tweet message = (Tweet) msg.poll();
-		assertEquals(2000, message.getId());
+		Message message = source.receive();
+		Tweet tweet = (Tweet) message.getPayload();
+		assertEquals(2000, tweet.getId());
 		source.stop();
 		Thread.sleep(3000);
 		
@@ -190,10 +193,12 @@ public class DirectMessageReceivingMessageSourceTests {
 		Thread.sleep(1000);
 		msg = (Queue) TestUtils.getPropertyValue(source, "tweets");
 		assertTrue(!CollectionUtils.isEmpty(msg));	
-		message = (Tweet) msg.poll();
-		assertEquals(3000, message.getId());
-		message = (Tweet) msg.poll();
-		assertEquals(4000, message.getId());
+		message = source.receive();
+		tweet = (Tweet) message.getPayload();
+		assertEquals(3000, tweet.getId());
+		message = source.receive();
+		tweet = (Tweet) message.getPayload();
+		assertEquals(4000, tweet.getId());
 		file.delete();
 	}
 
diff --git a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/TimelineUpdateReceivingMessageSourceTests.java b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/TimelineUpdateReceivingMessageSourceTests.java
index 25fec54fba..fb0dc8e224 100644
--- a/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/TimelineUpdateReceivingMessageSourceTests.java
+++ b/spring-integration-twitter/src/test/java/org/springframework/integration/twitter/inbound/TimelineUpdateReceivingMessageSourceTests.java
@@ -34,6 +34,7 @@ import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
 import org.springframework.beans.factory.support.DefaultListableBeanFactory;
+import org.springframework.integration.Message;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.store.PropertiesPersistingMetadataStore;
 import org.springframework.integration.test.util.TestUtils;
@@ -165,10 +166,11 @@ public class TimelineUpdateReceivingMessageSourceTests {
 		Queue msg = (Queue) TestUtils.getPropertyValue(source, "tweets");
 		assertTrue(!CollectionUtils.isEmpty(msg));	
 		assertEquals(1, msg.size()); // because the other message has a older timestamp and is assumed to be read by
-		Tweet message = (Tweet) msg.poll();
-		assertEquals(2000, message.getId());
+		Message message = (Message) source.receive();
+		Tweet tweet = (Tweet) message.getPayload();
+		assertEquals(2000, tweet.getId());
 		source.stop();
-		Thread.sleep(3000);
+		Thread.sleep(2000);
 		
 		
 		// Resuming
@@ -190,10 +192,12 @@ public class TimelineUpdateReceivingMessageSourceTests {
 		Thread.sleep(1000);
 		msg = (Queue) TestUtils.getPropertyValue(source, "tweets");
 		assertTrue(!CollectionUtils.isEmpty(msg));	
-		message = (Tweet) msg.poll();
-		assertEquals(3000, message.getId());
-		message = (Tweet) msg.poll();
-		assertEquals(4000, message.getId());
+		message = (Message) source.receive();
+		tweet = (Tweet) message.getPayload();
+		assertEquals(3000, tweet.getId());
+		message = (Message) source.receive();
+		tweet = (Tweet) message.getPayload();
+		assertEquals(4000, tweet.getId());
 		file.delete();
 	}