superdiaodiao commented on code in PR #57297:
URL: https://github.com/apache/spark/pull/57297#discussion_r3595808138
##########
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 {
+ logWarning(log"Polled zero visible records for ${MDC(TOPIC_PARTITION,
topicPartition)} " +
Review Comment:
Thanks for the review. That makes sense for this valid path; I lowered it to
DEBUG and updated the PR description accordingly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
