diff --git a/connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java b/connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java
index 93ab2bdd96474..5758f15e9d5b2 100644
--- a/connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java
+++ b/connect/api/src/main/java/org/apache/kafka/connect/source/SourceTask.java
@@ -46,7 +46,7 @@ public void initialize(SourceTaskContext context) {
     /**
      * <p>
      * Poll this source task for new records. If no data is currently available, this method
-     * should block but return control to the caller regularly (by returning {@code null}) in
+     * should block but return control to the caller regularly (by returning {@code null} or empty list) in
      * order for the task to transition to the {@code PAUSED} state if requested to do so.
      * </p>
      * <p>
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
index da2f9ef7c9245..fd4c758595ffc 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java
@@ -231,15 +231,15 @@ public void execute() {
 
                 maybeThrowProducerSendException();
 
-                if (toSend == null) {
+                if (toSend == null || toSend.isEmpty()) {
                     log.trace("{} Nothing to send to Kafka. Polling source for additional records", this);
                     long start = time.milliseconds();
                     toSend = poll();
-                    if (toSend != null) {
+                    if (toSend != null && !toSend.isEmpty()) {
                         recordPollReturned(toSend.size(), time.milliseconds() - start);
                     }
                 }
-                if (toSend == null)
+                if (toSend == null || toSend.isEmpty())
                     continue;
                 log.trace("{} About to send {} records to Kafka", this, toSend.size());
                 if (!sendRecords())