HeartSaVioR commented on code in PR #52745:
URL: https://github.com/apache/spark/pull/52745#discussion_r2480247179
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumer.scala:
##########
@@ -392,12 +392,22 @@ private[kafka010] class KafkaDataConsumer(
.getOrElse("")
val walTime = System.nanoTime() - startTimestampNano
+ // Get task context information for correlation with executor logs
+ val taskCtx = TaskContext.get()
+ val taskContextInfo = if (taskCtx != null) {
+ log" for taskId=${MDC(TASK_ATTEMPT_ID, taskCtx.taskAttemptId())} " +
+ log"partitionId=${MDC(PARTITION_ID, taskCtx.partitionId())}"
+ } else {
+ log""
+ }
+
logInfo(log"From Kafka ${MDC(CONSUMER, kafkaMeta)} read " +
log"${MDC(NUM_RECORDS_READ, totalRecordsRead)} records through " +
log"${MDC(NUM_KAFKA_PULLS, numPolls)} polls " +
log"(polled out ${MDC(NUM_KAFKA_RECORDS_PULLED, numRecordsPolled)}
records), " +
log"taking ${MDC(TOTAL_TIME_READ, totalTimeReadNanos /
NANOS_PER_MILLIS.toDouble)} ms, " +
- log"during time span of ${MDC(TIME, walTime /
NANOS_PER_MILLIS.toDouble)} ms."
+ log"during time span of ${MDC(TIME, walTime /
NANOS_PER_MILLIS.toDouble)} ms." +
Review Comment:
nit: `.` can be added in the end of string for `taskContextInfo`.
##########
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumerSuite.scala:
##########
Review Comment:
These tests does not actually verify the change - if we can't test the
behavior, let's not add tests.
--
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]
