This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new fe0e50253e3d [SPARK-49123][SS] Log empty Kafka polls with advanced
offsets
fe0e50253e3d is described below
commit fe0e50253e3d72f9317582659e096c36230b9ae9
Author: Chester <[email protected]>
AuthorDate: Fri Jul 17 17:51:43 2026 +0800
[SPARK-49123][SS] Log empty Kafka polls with advanced offsets
### What changes were proposed in this pull request?
This PR updates `KafkaDataConsumer` to log a debug message when a Kafka
poll returns zero visible records but advances the consumer position. It also
fixes the nearby comment from "not invisible" to "invisible".
### Why are the changes needed?
When transactional or aborted Kafka messages are invisible under
`isolation.level=read_committed`, Spark can poll zero visible records while the
offset still advances. This is a valid path, so Spark should not throw, but the
current behavior has little diagnostic context when debug logging is enabled.
The debug message gives developers the group id, topic partition, poll
timeout, and offset movement so they can distinguish this path from
out-of-range offsets and true poll timeouts while troubleshooting.
### Does this PR introduce _any_ user-facing change?
No. This only adjusts debug-level diagnostics and a code comment.
### How was this patch tested?
- `git diff --check`
- `python3 dev/structured_logging_style.py --help` (the script completed
and reported the structured logging style check passed)
Not run locally:
- `./dev/scalafmt --test` was attempted, but the script maps to
`./build/mvn scalafmt:format ...` and produced no output for over 90 seconds,
so I interrupted it.
- `./build/mvn -pl connector/kafka-0-10-sql -DskipTests compile` was
attempted, but produced no output for over 60 seconds, so I interrupted it.
### Was this patch authored or co-authored using generative AI tooling?
Yes, assisted by Codex.
Closes #57297 from superdiaodiao/codex/spark-49123-kafka-empty-poll-log.
Lead-authored-by: Chester <[email protected]>
Co-authored-by: xuyidiao <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
---
.../spark/sql/kafka010/consumer/KafkaDataConsumer.scala | 11 +++++++++--
1 file changed, 9 insertions(+), 2 deletions(-)
diff --git
a/connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumer.scala
b/connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumer.scala
index 126434625a8d..203fe24acc04 100644
---
a/connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumer.scala
+++
b/connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumer.scala
@@ -97,14 +97,21 @@ private[kafka010] class InternalKafkaConsumer(
// - `offset` is out of range so that Kafka returns nothing.
`OffsetOutOfRangeException` will
// be thrown.
// - Cannot fetch any data before timeout. `TimeoutException` will be
thrown.
- // - Fetched something but all of them are not invisible. This is a
valid case and let the
- // caller handles this.
+ // - Fetched something but all of them are invisible. This is a valid
case and let the
+ // caller handle this.
if (offset < range.earliest || offset >= range.latest) {
throw new OffsetOutOfRangeException(
Map(topicPartition -> java.lang.Long.valueOf(offset)).asJava)
} else if (offset == offsetAfterPoll) {
throw new TimeoutException(
s"Cannot fetch record for offset $offset in $pollTimeoutMs
milliseconds")
+ } else {
+ logDebug(log"Polled zero visible records for ${MDC(TOPIC_PARTITION,
topicPartition)} " +
+ log"in group ${MDC(GROUP_ID, groupId)} after ${MDC(TIMEOUT,
pollTimeoutMs)} " +
+ log"milliseconds, while the offset changed from ${MDC(OFFSET,
offset)} to " +
+ log"${MDC(UNTIL_OFFSET, offsetAfterPoll)}. This can happen when
fetched records are " +
+ log"invisible, for example transaction markers or aborted messages
with " +
+ log"`isolation.level=read_committed`.")
}
}
fetchedData
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
