diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
index 4e84f335eb..1e9f3ed616 100644
--- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
+++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java
@@ -139,6 +139,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 
 	private Duration closeTimeout = Duration.ofSeconds(DEFAULT_CLOSE_TIMEOUT);
 
+	public boolean newAssignment;
+
 	private volatile Consumer<K, V> consumer;
 
 	private volatile boolean pausing;
@@ -147,6 +149,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 
 	private volatile Iterator<ConsumerRecord<K, V>> recordsIterator;
 
+	private volatile boolean stopped;
+
 	/**
 	 * Construct an instance with the supplied parameters. Fetching multiple
 	 * records per poll will be disabled.
@@ -387,12 +391,14 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 	@Override
 	public synchronized void start() {
 		this.running = true;
+		this.stopped = false;
 	}
 
 	@Override
 	public synchronized void stop() {
 		stopConsumer();
 		this.running = false;
+		this.stopped = true;
 	}
 
 	@Override
@@ -412,6 +418,10 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 
 	@Override
 	protected synchronized Object doReceive() {
+		if (this.stopped) {
+			this.logger.debug("Message source is stopped; no records will be returned");
+			return null;
+		}
 		if (this.consumer == null) {
 			createConsumer();
 			this.running = true;
@@ -512,14 +522,27 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 		}
 		else {
 			synchronized (this.consumerMonitor) {
-				ConsumerRecords<K, V> records = this.consumer
-						.poll(this.assignedPartitions.isEmpty() ? this.assignTimeout : this.pollTimeout);
-				if (records == null || records.count() == 0) {
+				try {
+					ConsumerRecords<K, V> records = this.consumer
+							.poll(this.assignedPartitions.isEmpty() ? this.assignTimeout : this.pollTimeout);
+					this.logger.debug(() -> records == null
+							? "Received null"
+							: "Received " + records.count() + " records");
+					if (records == null || records.count() == 0) {
+						return null;
+					}
+					this.remainingCount.set(records.count());
+					this.recordsIterator = records.iterator();
+					return nextRecord();
+				}
+				catch (WakeupException ex) {
+					this.logger.debug("Woken");
+					if (this.newAssignment) {
+						this.newAssignment = false;
+						return pollRecord();
+					}
 					return null;
 				}
-				this.remainingCount.set(records.count());
-				this.recordsIterator = records.iterator();
-				return nextRecord();
 			}
 		}
 	}
@@ -637,6 +660,8 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> impl
 					this.providedRebalanceListener.onPartitionsAssigned(partitions);
 				}
 			}
+			KafkaMessageSource.this.consumer.wakeup();
+			KafkaMessageSource.this.newAssignment = true;
 		}
 
 	}