You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ge...@apache.org on 2024/05/04 06:33:58 UTC
(spark) branch master updated: [SPARK-47578][CORE] Migrate logWarning with variables to structured logging framework
This is an automated email from the ASF dual-hosted git repository.
gengliang 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 111529e0dea6 [SPARK-47578][CORE] Migrate logWarning with variables to structured logging framework
111529e0dea6 is described below
commit 111529e0dea68bf5343cc6aabba53b59e5d21830
Author: Daniel Tenedorio <da...@databricks.com>
AuthorDate: Fri May 3 23:33:46 2024 -0700
[SPARK-47578][CORE] Migrate logWarning with variables to structured logging framework
### What changes were proposed in this pull request?
Migrate logWarning with variables of the Spark Core module to structured logging framework. This transforms the logWarning entries of the following API
```
def logWarning(msg: => String): Unit
```
to
```
def logWarning(entry: LogEntry): Unit
```
### Why are the changes needed?
To enhance Apache Spark's logging system by implementing structured logging.
### Does this PR introduce _any_ user-facing change?
Yes, Spark core logs will contain additional MDC
### How was this patch tested?
Compiler and scala style checks, as well as code review.
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #46309 from dtenedor/spark-core-log-warn.
Authored-by: Daniel Tenedorio <da...@databricks.com>
Signed-off-by: Gengliang Wang <ge...@apache.org>
---
.../scala/org/apache/spark/internal/LogKey.scala | 97 ++++++++++++++++++++++
.../main/scala/org/apache/spark/Dependency.scala | 14 ++--
.../apache/spark/ExecutorAllocationManager.scala | 17 ++--
.../scala/org/apache/spark/HeartbeatReceiver.scala | 12 ++-
.../scala/org/apache/spark/MapOutputTracker.scala | 9 +-
.../main/scala/org/apache/spark/SparkConf.scala | 32 ++++---
.../main/scala/org/apache/spark/SparkContext.scala | 38 +++++----
.../src/main/scala/org/apache/spark/SparkEnv.scala | 6 +-
.../org/apache/spark/api/python/PythonRunner.scala | 6 +-
.../spark/api/python/PythonWorkerFactory.scala | 6 +-
.../org/apache/spark/api/python/SerDeUtil.scala | 31 ++++---
.../org/apache/spark/api/r/RBackendHandler.scala | 16 ++--
.../scala/org/apache/spark/deploy/Client.scala | 5 +-
.../apache/spark/deploy/DriverTimeoutPlugin.scala | 8 +-
.../org/apache/spark/deploy/SparkSubmit.scala | 18 ++--
.../apache/spark/deploy/SparkSubmitArguments.scala | 5 +-
.../spark/deploy/client/StandaloneAppClient.scala | 15 ++--
.../spark/deploy/history/ApplicationCache.scala | 5 +-
.../deploy/history/EventLogFileCompactor.scala | 5 +-
.../spark/deploy/history/EventLogFileWriters.scala | 11 ++-
.../spark/deploy/history/FsHistoryProvider.scala | 31 ++++---
.../deploy/history/HistoryServerDiskManager.scala | 11 ++-
.../master/FileSystemPersistenceEngine.scala | 5 +-
.../org/apache/spark/deploy/master/Master.scala | 80 ++++++++++--------
.../spark/deploy/rest/RestSubmissionClient.scala | 4 +-
.../security/HBaseDelegationTokenProvider.scala | 2 +-
.../security/HadoopDelegationTokenManager.scala | 11 ++-
.../security/HadoopFSDelegationTokenProvider.scala | 22 +++--
.../apache/spark/deploy/worker/DriverRunner.scala | 7 +-
.../spark/deploy/worker/ExecutorRunner.scala | 9 +-
.../org/apache/spark/deploy/worker/Worker.scala | 41 +++++----
.../executor/CoarseGrainedExecutorBackend.scala | 24 +++---
.../scala/org/apache/spark/executor/Executor.scala | 34 ++++----
.../io/HadoopMapReduceCommitProtocol.scala | 13 +--
.../spark/internal/plugin/PluginEndpoint.scala | 13 +--
.../apache/spark/memory/ExecutionMemoryPool.scala | 8 +-
.../apache/spark/memory/StorageMemoryPool.scala | 7 +-
.../apache/spark/metrics/ExecutorMetricType.scala | 14 ++--
.../org/apache/spark/metrics/MetricsSystem.scala | 17 ++--
.../spark/network/netty/NettyBlockRpcServer.scala | 19 +++--
.../scala/org/apache/spark/rdd/HadoopRDD.scala | 27 +++---
.../scala/org/apache/spark/rdd/NewHadoopRDD.scala | 27 +++---
.../org/apache/spark/rdd/PairRDDFunctions.scala | 12 +--
core/src/main/scala/org/apache/spark/rdd/RDD.scala | 6 +-
.../apache/spark/rdd/ReliableCheckpointRDD.scala | 12 +--
.../apache/spark/resource/ResourceProfile.scala | 7 +-
.../org/apache/spark/resource/ResourceUtils.scala | 33 ++++----
.../org/apache/spark/rpc/netty/Dispatcher.scala | 9 +-
.../scala/org/apache/spark/rpc/netty/Inbox.scala | 5 +-
.../org/apache/spark/rpc/netty/NettyRpcEnv.scala | 7 +-
.../scala/org/apache/spark/rpc/netty/Outbox.scala | 2 +-
.../apache/spark/scheduler/AsyncEventQueue.scala | 7 +-
.../org/apache/spark/scheduler/DAGScheduler.scala | 39 +++++----
.../org/apache/spark/scheduler/HealthTracker.scala | 13 +--
.../apache/spark/scheduler/ReplayListenerBus.scala | 11 +--
.../spark/scheduler/SchedulableBuilder.scala | 38 +++++----
.../apache/spark/scheduler/TaskSchedulerImpl.scala | 23 +++--
.../apache/spark/scheduler/TaskSetManager.scala | 23 +++--
.../cluster/CoarseGrainedSchedulerBackend.scala | 16 ++--
.../cluster/StandaloneSchedulerBackend.scala | 4 +-
.../apache/spark/security/CryptoStreamUtils.scala | 7 +-
.../spark/shuffle/IndexShuffleBlockResolver.scala | 18 ++--
.../apache/spark/shuffle/ShuffleBlockPusher.scala | 15 ++--
.../scala/org/apache/spark/status/KVUtils.scala | 5 +-
.../org/apache/spark/storage/BlockManager.scala | 51 +++++++-----
.../spark/storage/BlockManagerDecommissioner.scala | 20 +++--
.../apache/spark/storage/BlockManagerMaster.scala | 14 ++--
.../spark/storage/BlockManagerMasterEndpoint.scala | 14 ++--
.../spark/storage/BlockReplicationPolicy.scala | 6 +-
.../apache/spark/storage/DiskBlockManager.scala | 4 +-
.../spark/storage/DiskBlockObjectWriter.scala | 4 +-
.../scala/org/apache/spark/storage/DiskStore.scala | 7 +-
.../org/apache/spark/storage/FallbackStorage.scala | 7 +-
.../spark/storage/PushBasedFetchHelper.scala | 14 ++--
.../storage/ShuffleBlockFetcherIterator.scala | 26 +++---
.../org/apache/spark/storage/TopologyMapper.scala | 5 +-
.../apache/spark/storage/memory/MemoryStore.scala | 26 +++---
.../scala/org/apache/spark/ui/JettyUtils.scala | 7 +-
.../apache/spark/ui/scope/RDDOperationGraph.scala | 6 +-
.../org/apache/spark/util/AccumulatorV2.scala | 6 +-
.../org/apache/spark/util/DependencyUtils.scala | 8 +-
.../org/apache/spark/util/HadoopFSUtils.scala | 13 +--
.../apache/spark/util/PeriodicCheckpointer.scala | 9 +-
.../scala/org/apache/spark/util/SignalUtils.scala | 7 +-
.../org/apache/spark/util/SizeEstimator.scala | 6 +-
.../main/scala/org/apache/spark/util/Utils.scala | 60 +++++++------
.../util/collection/ExternalAppendOnlyMap.scala | 5 +-
.../apache/spark/util/logging/DriverLogger.scala | 7 +-
.../apache/spark/util/logging/FileAppender.scala | 10 ++-
.../spark/util/logging/RollingFileAppender.scala | 9 +-
.../apache/spark/util/logging/RollingPolicy.scala | 13 +--
91 files changed, 911 insertions(+), 547 deletions(-)
diff --git a/common/utils/src/main/scala/org/apache/spark/internal/LogKey.scala b/common/utils/src/main/scala/org/apache/spark/internal/LogKey.scala
index 238432d354f6..d4e1d9f535af 100644
--- a/common/utils/src/main/scala/org/apache/spark/internal/LogKey.scala
+++ b/common/utils/src/main/scala/org/apache/spark/internal/LogKey.scala
@@ -48,7 +48,10 @@ object LogKeys {
case object BATCH_TIMESTAMP extends LogKey
case object BATCH_WRITE extends LogKey
case object BLOCK_ID extends LogKey
+ case object BLOCK_IDS extends LogKey
case object BLOCK_MANAGER_ID extends LogKey
+ case object BLOCK_MANAGER_IDS extends LogKey
+ case object BLOCK_TYPE extends LogKey
case object BOOT extends LogKey
case object BROADCAST extends LogKey
case object BROADCAST_ID extends LogKey
@@ -70,6 +73,7 @@ object LogKeys {
case object CHOSEN_WATERMARK extends LogKey
case object CLASS_LOADER extends LogKey
case object CLASS_NAME extends LogKey
+ case object CLASS_PATH extends LogKey
case object CLAUSES extends LogKey
case object CLUSTER_CENTROIDS extends LogKey
case object CLUSTER_ID extends LogKey
@@ -94,12 +98,18 @@ object LogKeys {
case object CONFIG3 extends LogKey
case object CONFIG4 extends LogKey
case object CONFIG5 extends LogKey
+ case object CONFIG_DEPRECATION_MESSAGE extends LogKey
+ case object CONFIG_KEY_UPDATED extends LogKey
+ case object CONFIG_VERSION extends LogKey
case object CONSUMER extends LogKey
case object CONTAINER extends LogKey
case object CONTAINER_ID extends LogKey
case object CONTAINER_STATE extends LogKey
+ case object CONTEXT extends LogKey
+ case object CONTEXT_CREATION_SITE extends LogKey
case object COST extends LogKey
case object COUNT extends LogKey
+ case object CREATED_POOL_NAME extends LogKey
case object CROSS_VALIDATION_METRIC extends LogKey
case object CROSS_VALIDATION_METRICS extends LogKey
case object CSV_HEADER_COLUMN_NAME extends LogKey
@@ -110,7 +120,9 @@ object LogKeys {
case object CSV_SOURCE extends LogKey
case object CURRENT_BATCH_ID extends LogKey
case object CURRENT_FILE extends LogKey
+ case object CURRENT_MEMORY_BYTES extends LogKey
case object CURRENT_PATH extends LogKey
+ case object CURRENT_TIME extends LogKey
case object DATA extends LogKey
case object DATABASE_NAME extends LogKey
case object DATAFRAME_CACHE_ENTRY extends LogKey
@@ -121,16 +133,20 @@ object LogKeys {
case object DEFAULT_COMPACTION_INTERVAL extends LogKey
case object DEFAULT_COMPACT_INTERVAL extends LogKey
case object DEFAULT_ISOLATION_LEVEL extends LogKey
+ case object DEFAULT_NAME extends LogKey
+ case object DEFAULT_SCHEDULING_MODE extends LogKey
case object DEFAULT_VALUE extends LogKey
case object DELAY extends LogKey
case object DELEGATE extends LogKey
case object DELTA extends LogKey
+ case object DEPRECATED_KEY extends LogKey
case object DESCRIPTION extends LogKey
case object DESIRED_NUM_PARTITIONS extends LogKey
case object DFS_FILE extends LogKey
case object DIFF_DELTA extends LogKey
case object DIVISIBLE_CLUSTER_INDICES_SIZE extends LogKey
case object DRIVER_ID extends LogKey
+ case object DRIVER_LIBRARY_PATH_KEY extends LogKey
case object DROPPED_PARTITIONS extends LogKey
case object DURATION extends LogKey
case object EARLIEST_LOADED_VERSION extends LogKey
@@ -145,12 +161,15 @@ object LogKeys {
case object ERROR extends LogKey
case object ESTIMATOR_PARAMETER_MAP extends LogKey
case object EVALUATED_FILTERS extends LogKey
+ case object EVENT_LOG_DESTINATION extends LogKey
case object EVENT_LOOP extends LogKey
+ case object EVENT_NAME extends LogKey
case object EVENT_QUEUE extends LogKey
case object EXCEPTION extends LogKey
case object EXECUTE_INFO extends LogKey
case object EXECUTE_KEY extends LogKey
case object EXECUTION_PLAN_LEAVES extends LogKey
+ case object EXECUTOR_BACKEND extends LogKey
case object EXECUTOR_DESIRED_COUNT extends LogKey
case object EXECUTOR_ENVS extends LogKey
case object EXECUTOR_ENV_REGEX extends LogKey
@@ -161,7 +180,10 @@ object LogKeys {
case object EXECUTOR_RESOURCES extends LogKey
case object EXECUTOR_STATE extends LogKey
case object EXECUTOR_TARGET_COUNT extends LogKey
+ case object EXECUTOR_TIMEOUT extends LogKey
+ case object EXEC_AMOUNT extends LogKey
case object EXISTING_FILE extends LogKey
+ case object EXISTING_PATH extends LogKey
case object EXIT_CODE extends LogKey
case object EXPECTED_NUM_FILES extends LogKey
case object EXPECTED_PARTITION_COLUMN extends LogKey
@@ -181,10 +203,13 @@ object LogKeys {
case object FILE_LENGTH_XATTR extends LogKey
case object FILE_MODIFICATION_TIME extends LogKey
case object FILE_NAME extends LogKey
+ case object FILE_NAME2 extends LogKey
+ case object FILE_NAME3 extends LogKey
case object FILE_START_OFFSET extends LogKey
case object FILE_VERSION extends LogKey
case object FILTER extends LogKey
case object FILTERS extends LogKey
+ case object FINAL_CONTEXT extends LogKey
case object FINAL_PATH extends LogKey
case object FINISH_TRIGGER_DURATION extends LogKey
case object FROM_OFFSET extends LogKey
@@ -197,6 +222,8 @@ object LogKeys {
case object HADOOP_VERSION extends LogKey
case object HASH_JOIN_KEYS extends LogKey
case object HASH_MAP_SIZE extends LogKey
+ case object HAS_R_PACKAGE extends LogKey
+ case object HEARTBEAT extends LogKey
case object HEARTBEAT_INTERVAL extends LogKey
case object HISTORY_DIR extends LogKey
case object HIVE_CLIENT_VERSION extends LogKey
@@ -205,7 +232,9 @@ object LogKeys {
case object HIVE_OPERATION_TYPE extends LogKey
case object HOST extends LogKey
case object HOST_PORT extends LogKey
+ case object HOST_PORT2 extends LogKey
case object HUGE_METHOD_LIMIT extends LogKey
+ case object IDENTIFIER extends LogKey
case object INCOMPATIBLE_TYPES extends LogKey
case object INDEX extends LogKey
case object INDEX_FILE_NUM extends LogKey
@@ -218,6 +247,10 @@ object LogKeys {
case object INPUT extends LogKey
case object INTERVAL extends LogKey
case object ISOLATION_LEVEL extends LogKey
+ case object ISSUE_DATE extends LogKey
+ case object IS_NETWORK_REQUEST_DONE extends LogKey
+ case object JAR_ENTRY extends LogKey
+ case object JAR_MESSAGE extends LogKey
case object JOB_ID extends LogKey
case object JOIN_CONDITION extends LogKey
case object JOIN_CONDITION_SUB_EXPR extends LogKey
@@ -242,6 +275,7 @@ object LogKeys {
case object LOADED_VERSION extends LogKey
case object LOAD_FACTOR extends LogKey
case object LOAD_TIME extends LogKey
+ case object LOCATION extends LogKey
case object LOGICAL_PLAN_COLUMNS extends LogKey
case object LOGICAL_PLAN_LEAVES extends LogKey
case object LOG_ID extends LogKey
@@ -249,6 +283,7 @@ object LogKeys {
case object LOG_TYPE extends LogKey
case object LOWER_BOUND extends LogKey
case object MALFORMATTED_STIRNG extends LogKey
+ case object MAP_ID extends LogKey
case object MASTER_URL extends LogKey
case object MAX_ATTEMPTS extends LogKey
case object MAX_CACHE_UNTIL_HIGHEST_CONSUMED_SIZE extends LogKey
@@ -258,42 +293,61 @@ object LogKeys {
case object MAX_EXECUTOR_FAILURES extends LogKey
case object MAX_FILE_VERSION extends LogKey
case object MAX_JVM_METHOD_PARAMS_LENGTH extends LogKey
+ case object MAX_LOG_NUM_POLICY extends LogKey
case object MAX_MEMORY_SIZE extends LogKey
case object MAX_METHOD_CODE_SIZE extends LogKey
case object MAX_NUM_PARTITIONS extends LogKey
case object MAX_NUM_ROWS_IN_MEMORY_BUFFER extends LogKey
case object MAX_SIZE extends LogKey
+ case object MAX_SLOTS extends LogKey
case object MAX_SPLIT_BYTES extends LogKey
case object MAX_TABLE_PARTITION_METADATA_SIZE extends LogKey
+ case object MEMORY_POOL_NAME extends LogKey
case object MEMORY_SIZE extends LogKey
case object MERGE_DIR_NAME extends LogKey
case object MESSAGE extends LogKey
case object METADATA_DIRECTORY extends LogKey
case object METADATA_JSON extends LogKey
case object METHOD_NAME extends LogKey
+ case object METHOD_PARAMETER_TYPES extends LogKey
case object METRICS_JSON extends LogKey
case object MIN_COMPACTION_BATCH_ID extends LogKey
case object MIN_FREQUENT_PATTERN_COUNT extends LogKey
case object MIN_POINT_PER_CLUSTER extends LogKey
+ case object MIN_SHARE extends LogKey
case object MIN_SIZE extends LogKey
+ case object MIN_TIME extends LogKey
case object MIN_VERSION_NUMBER extends LogKey
case object MODEL_WEIGHTS extends LogKey
case object NAMESPACE extends LogKey
+ case object NETWORK_IF extends LogKey
case object NEW_FEATURE_COLUMN_NAME extends LogKey
case object NEW_LABEL_COLUMN_NAME extends LogKey
case object NEW_PATH extends LogKey
+ case object NEW_STATE extends LogKey
case object NEW_VALUE extends LogKey
case object NODES extends LogKey
case object NODE_LOCATION extends LogKey
+ case object NON_BUILT_IN_CONNECTORS extends LogKey
case object NORM extends LogKey
case object NUM_ADDED_PARTITIONS extends LogKey
case object NUM_BIN extends LogKey
case object NUM_BYTES extends LogKey
+ case object NUM_BYTES_CURRENT extends LogKey
+ case object NUM_BYTES_MAX extends LogKey
+ case object NUM_BYTES_TO_WARN extends LogKey
+ case object NUM_BYTES_USED extends LogKey
case object NUM_CLASSES extends LogKey
case object NUM_COLUMNS extends LogKey
case object NUM_CONCURRENT_WRITER extends LogKey
+ case object NUM_CORES extends LogKey
case object NUM_DROPPED_PARTITIONS extends LogKey
+ case object NUM_EVENTS extends LogKey
case object NUM_EXAMPLES extends LogKey
+ case object NUM_EXECUTOR_CORES extends LogKey
+ case object NUM_EXECUTOR_CORES_REMAINING extends LogKey
+ case object NUM_EXECUTOR_CORES_TOTAL extends LogKey
+ case object NUM_FAILURES extends LogKey
case object NUM_FEATURES extends LogKey
case object NUM_FILES extends LogKey
case object NUM_FILES_COPIED extends LogKey
@@ -305,12 +359,24 @@ object LogKeys {
case object NUM_LOADED_ENTRIES extends LogKey
case object NUM_LOCAL_FREQUENT_PATTERN extends LogKey
case object NUM_PARTITIONS extends LogKey
+ case object NUM_PARTITIONS2 extends LogKey
case object NUM_PARTITION_VALUES extends LogKey
+ case object NUM_PEERS extends LogKey
+ case object NUM_PEERS_REPLICATED_TO extends LogKey
+ case object NUM_PEERS_TO_REPLICATE_TO extends LogKey
+ case object NUM_PENDING_LAUNCH_TASKS extends LogKey
case object NUM_POINT extends LogKey
case object NUM_PREFIXES extends LogKey
case object NUM_PRUNED extends LogKey
+ case object NUM_REPLICAS extends LogKey
+ case object NUM_REQUESTS extends LogKey
+ case object NUM_RESOURCE_SLOTS extends LogKey
+ case object NUM_RETRIES extends LogKey
case object NUM_RIGHT_PARTITION_VALUES extends LogKey
case object NUM_SEQUENCES extends LogKey
+ case object NUM_SLOTS extends LogKey
+ case object NUM_TASKS extends LogKey
+ case object NUM_TASK_CPUS extends LogKey
case object NUM_VERSIONS_RETAIN extends LogKey
case object OBJECT_AGG_SORT_BASED_FALLBACK_THRESHOLD extends LogKey
case object OBJECT_ID extends LogKey
@@ -318,6 +384,7 @@ object LogKeys {
case object OFFSETS extends LogKey
case object OFFSET_SEQUENCE_METADATA extends LogKey
case object OLD_BLOCK_MANAGER_ID extends LogKey
+ case object OLD_GENERATION_GC extends LogKey
case object OLD_VALUE extends LogKey
case object OPEN_COST_IN_BYTES extends LogKey
case object OPTIMIZED_PLAN_COLUMNS extends LogKey
@@ -329,13 +396,17 @@ object LogKeys {
case object OVERHEAD_MEMORY_SIZE extends LogKey
case object PARSE_MODE extends LogKey
case object PARTITIONED_FILE_READER extends LogKey
+ case object PARTITIONER extends LogKey
+ case object PARTITIONS_SIZE extends LogKey
case object PARTITION_ID extends LogKey
case object PARTITION_SPECIFICATION extends LogKey
case object PARTITION_SPECS extends LogKey
case object PATH extends LogKey
case object PATHS extends LogKey
+ case object PEER extends LogKey
case object PERCENT extends LogKey
case object PIPELINE_STAGE_UID extends LogKey
+ case object PLUGIN_NAME extends LogKey
case object POD_COUNT extends LogKey
case object POD_ID extends LogKey
case object POD_NAME extends LogKey
@@ -345,14 +416,18 @@ object LogKeys {
case object POD_STATE extends LogKey
case object POD_TARGET_COUNT extends LogKey
case object POLICY extends LogKey
+ case object POOL_NAME extends LogKey
case object PORT extends LogKey
+ case object PORT2 extends LogKey
case object POST_SCAN_FILTERS extends LogKey
case object PREDICATE extends LogKey
case object PREDICATES extends LogKey
case object PRETTY_ID_STRING extends LogKey
case object PRINCIPAL extends LogKey
+ case object PROCESS extends LogKey
case object PROCESSING_TIME extends LogKey
case object PRODUCER_ID extends LogKey
+ case object PROPERTY_NAME extends LogKey
case object PROVIDER extends LogKey
case object PUSHED_FILTERS extends LogKey
case object PVC_METADATA_NAME extends LogKey
@@ -366,6 +441,7 @@ object LogKeys {
case object QUERY_PLAN_LENGTH_MAX extends LogKey
case object QUERY_RUN_ID extends LogKey
case object RANGE extends LogKey
+ case object RDD_DESCRIPTION extends LogKey
case object RDD_ID extends LogKey
case object READ_LIMIT extends LogKey
case object REASON extends LogKey
@@ -381,11 +457,16 @@ object LogKeys {
case object RELATION_OUTPUT extends LogKey
case object RELATIVE_TOLERANCE extends LogKey
case object REMAINING_PARTITIONS extends LogKey
+ case object REMOTE_ADDRESS extends LogKey
+ case object REMOVE_FROM_MASTER extends LogKey
case object REPORT_DETAILS extends LogKey
case object RESOURCE extends LogKey
case object RESOURCE_NAME extends LogKey
case object RESOURCE_PROFILE_ID extends LogKey
case object RESOURCE_PROFILE_IDS extends LogKey
+ case object RESOURCE_PROFILE_TO_TOTAL_EXECS extends LogKey
+ case object RESULT_SIZE_BYTES extends LogKey
+ case object RESULT_SIZE_BYTES_MAX extends LogKey
case object RETRY_COUNT extends LogKey
case object RETRY_INTERVAL extends LogKey
case object RIGHT_EXPR extends LogKey
@@ -394,9 +475,12 @@ object LogKeys {
case object ROCKS_DB_LOG_LEVEL extends LogKey
case object ROCKS_DB_LOG_MESSAGE extends LogKey
case object RPC_ENDPOINT_REF extends LogKey
+ case object RPC_MESSAGE_CAPACITY extends LogKey
+ case object RULE_BATCH_NAME extends LogKey
case object RULE_NAME extends LogKey
case object RULE_NUMBER_OF_RUNS extends LogKey
case object RUN_ID extends LogKey
+ case object SCHEDULER_POOL_NAME extends LogKey
case object SCHEMA extends LogKey
case object SCHEMA2 extends LogKey
case object SERVICE_NAME extends LogKey
@@ -404,15 +488,18 @@ object LogKeys {
case object SESSION_ID extends LogKey
case object SESSION_KEY extends LogKey
case object SHARD_ID extends LogKey
+ case object SHELL_COMMAND extends LogKey
case object SHUFFLE_BLOCK_INFO extends LogKey
case object SHUFFLE_ID extends LogKey
case object SHUFFLE_MERGE_ID extends LogKey
case object SHUFFLE_SERVICE_NAME extends LogKey
+ case object SIGNAL extends LogKey
case object SIZE extends LogKey
case object SLEEP_TIME extends LogKey
case object SLIDE_DURATION extends LogKey
case object SMALLEST_CLUSTER_INDEX extends LogKey
case object SNAPSHOT_VERSION extends LogKey
+ case object SOCKET_ADDRESS extends LogKey
case object SPARK_DATA_STREAM extends LogKey
case object SPARK_PLAN_ID extends LogKey
case object SQL_TEXT extends LogKey
@@ -449,6 +536,7 @@ object LogKeys {
case object TASK_ATTEMPT_ID extends LogKey
case object TASK_ID extends LogKey
case object TASK_NAME extends LogKey
+ case object TASK_REQUIREMENTS extends LogKey
case object TASK_SET_NAME extends LogKey
case object TASK_STATE extends LogKey
case object TEMP_FILE extends LogKey
@@ -462,6 +550,7 @@ object LogKeys {
case object TIMER extends LogKey
case object TIME_UNITS extends LogKey
case object TIP extends LogKey
+ case object TOKEN_KIND extends LogKey
case object TOKEN_REGEX extends LogKey
case object TOPIC extends LogKey
case object TOPIC_PARTITION extends LogKey
@@ -488,6 +577,7 @@ object LogKeys {
case object UNTIL_OFFSET extends LogKey
case object UPPER_BOUND extends LogKey
case object URI extends LogKey
+ case object URIS extends LogKey
case object USER_ID extends LogKey
case object USER_NAME extends LogKey
case object UUID extends LogKey
@@ -499,9 +589,16 @@ object LogKeys {
case object WAIT_SEND_TIME extends LogKey
case object WAIT_TIME extends LogKey
case object WATERMARK_CONSTRAINT extends LogKey
+ case object WEIGHT extends LogKey
case object WEIGHTED_NUM extends LogKey
+ case object WORKER extends LogKey
+ case object WORKER_HOST extends LogKey
+ case object WORKER_ID extends LogKey
+ case object WORKER_PORT extends LogKey
case object WORKER_URL extends LogKey
case object WRITE_AHEAD_LOG_INFO extends LogKey
case object WRITE_JOB_UUID extends LogKey
+ case object XML_SCHEDULING_MODE extends LogKey
case object XSD_PATH extends LogKey
+ case object YOUNG_GENERATION_GC extends LogKey
}
diff --git a/core/src/main/scala/org/apache/spark/Dependency.scala b/core/src/main/scala/org/apache/spark/Dependency.scala
index 3b5bb6792928..3b7c7778e26c 100644
--- a/core/src/main/scala/org/apache/spark/Dependency.scala
+++ b/core/src/main/scala/org/apache/spark/Dependency.scala
@@ -24,7 +24,8 @@ import scala.reflect.ClassTag
import org.roaringbitmap.RoaringBitmap
import org.apache.spark.annotation.DeveloperApi
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.rdd.RDD
import org.apache.spark.serializer.Serializer
import org.apache.spark.shuffle.{ShuffleHandle, ShuffleWriteProcessor}
@@ -211,10 +212,13 @@ class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](
// This may crash the driver with an OOM error.
if (numPartitions.toLong * partitioner.numPartitions.toLong > (1L << 30)) {
logWarning(
- s"The number of shuffle blocks (${numPartitions.toLong * partitioner.numPartitions.toLong})" +
- s" for shuffleId ${shuffleId} for ${_rdd} with ${numPartitions} partitions" +
- " is possibly too large, which could cause the driver to crash with an out-of-memory" +
- " error. Consider decreasing the number of partitions in this shuffle stage."
+ log"The number of shuffle blocks " +
+ log"(${MDC(NUM_PARTITIONS, numPartitions.toLong * partitioner.numPartitions.toLong)})" +
+ log" for shuffleId ${MDC(SHUFFLE_ID, shuffleId)} " +
+ log"for ${MDC(RDD_DESCRIPTION, _rdd)} " +
+ log"with ${MDC(NUM_PARTITIONS2, numPartitions)} partitions" +
+ log" is possibly too large, which could cause the driver to crash with an out-of-memory" +
+ log" error. Consider decreasing the number of partitions in this shuffle stage."
)
}
diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
index 94927caff1d7..d156d74b2316 100644
--- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
+++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
@@ -26,7 +26,8 @@ import scala.util.control.NonFatal
import com.codahale.metrics.{Counter, Gauge, MetricRegistry}
-import org.apache.spark.internal.{config, Logging}
+import org.apache.spark.internal.{config, Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.DECOMMISSION_ENABLED
import org.apache.spark.internal.config.Tests.TEST_DYNAMIC_ALLOCATION_SCHEDULE_ENABLED
@@ -542,8 +543,8 @@ private[spark] class ExecutorAllocationManager(
if (testing) {
throw new SparkException("ResourceProfile Id was UNKNOWN, this is not expected")
}
- logWarning(s"Not removing executor $executorIdToBeRemoved because the " +
- "ResourceProfile was UNKNOWN!")
+ logWarning(log"Not removing executor ${MDC(EXECUTOR_IDS, executorIdToBeRemoved)} " +
+ log"because the ResourceProfile was UNKNOWN!")
} else {
// get the running total as we remove or initialize it to the count - pendingRemoval
val newExecutorTotal = numExecutorsTotalPerRpId.getOrElseUpdate(rpId,
@@ -606,8 +607,9 @@ private[spark] class ExecutorAllocationManager(
logInfo(s"Executors ${executorsRemoved.mkString(",")} removed due to idle timeout.")
executorsRemoved.toSeq
} else {
- logWarning(s"Unable to reach the cluster manager to kill executor/s " +
- s"${executorIdsToBeRemoved.mkString(",")} or no executor eligible to kill!")
+ logWarning(log"Unable to reach the cluster manager to kill executor/s " +
+ log"${MDC(EXECUTOR_IDS, executorIdsToBeRemoved.mkString(","))} " +
+ log"or no executor eligible to kill!")
Seq.empty[String]
}
}
@@ -870,8 +872,9 @@ private[spark] class ExecutorAllocationManager(
// really complete and no tasks left
resourceProfileIdToStageAttempt(rpForStage.head) -= stageAttempt
} else {
- logWarning(s"Should have exactly one resource profile for stage $stageAttempt," +
- s" but have $rpForStage")
+ logWarning(log"Should have exactly one resource profile for stage " +
+ log"${MDC(STAGE_ATTEMPT, stageAttempt)}, but have " +
+ log"${MDC(RESOURCE_PROFILE_ID, rpForStage)}")
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala b/core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala
index 5999040894ae..92aea5959aab 100644
--- a/core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala
+++ b/core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala
@@ -23,7 +23,8 @@ import scala.collection.mutable.{HashMap, Map}
import scala.concurrent.Future
import org.apache.spark.executor.ExecutorMetrics
-import org.apache.spark.internal.{config, Logging}
+import org.apache.spark.internal.{config, Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.Network
import org.apache.spark.rpc.{IsolatedThreadSafeRpcEndpoint, RpcCallContext, RpcEnv}
import org.apache.spark.scheduler._
@@ -159,7 +160,8 @@ private[spark] class HeartbeatReceiver(sc: SparkContext, clock: Clock)
// Because Executor will sleep several seconds before sending the first "Heartbeat", this
// case rarely happens. However, if it really happens, log it and ask the executor to
// register itself again.
- logWarning(s"Dropping $heartbeat because TaskScheduler is not ready yet")
+ logWarning(log"Dropping ${MDC(HEARTBEAT, heartbeat)} " +
+ log"because TaskScheduler is not ready yet")
context.reply(HeartbeatResponse(reregisterBlockManager))
}
}
@@ -210,8 +212,10 @@ private[spark] class HeartbeatReceiver(sc: SparkContext, clock: Clock)
val now = clock.getTimeMillis()
for ((executorId, lastSeenMs) <- executorLastSeen) {
if (now - lastSeenMs > executorTimeoutMs) {
- logWarning(s"Removing executor $executorId with no recent heartbeats: " +
- s"${now - lastSeenMs} ms exceeds timeout $executorTimeoutMs ms")
+ logWarning(log"Removing executor ${MDC(EXECUTOR_ID, executorId)} " +
+ log"with no recent heartbeats: " +
+ log"${MDC(TIME_UNITS, now - lastSeenMs)} ms exceeds timeout " +
+ log"${MDC(EXECUTOR_TIMEOUT, executorTimeoutMs)} ms")
// Asynchronously kill the executor to avoid blocking the current thread
killExecutorThread.submit(new Runnable {
override def run(): Unit = Utils.tryLogNonFatalError {
diff --git a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
index a323dc9fcb47..6f3a354a2879 100644
--- a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
+++ b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
@@ -202,12 +202,14 @@ private class ShuffleStatus(
mapStatusesDeleted(index) = null
logInfo(s"Recover ${mapStatus.mapId} ${mapStatus.location}")
} else {
- logWarning(s"Asked to update map output ${mapId} for untracked map status.")
+ logWarning(log"Asked to update map output ${MDC(MAP_ID, mapId)} " +
+ log"for untracked map status.")
}
}
} catch {
case e: java.lang.NullPointerException =>
- logWarning(s"Unable to update map output for ${mapId}, status removed in-flight")
+ logWarning(log"Unable to update map output for ${MDC(MAP_ID, mapId)}, " +
+ log"status removed in-flight")
}
}
@@ -815,7 +817,8 @@ private[spark] class MapOutputTrackerMaster(
case Some(shuffleStatus) =>
shuffleStatus.updateMapOutput(mapId, bmAddress)
case None if shuffleMigrationEnabled =>
- logWarning(s"Asked to update map output for unknown shuffle ${shuffleId}")
+ logWarning(log"Asked to update map output for unknown shuffle " +
+ log"${MDC(SHUFFLE_ID, shuffleId)}")
case None =>
logError(log"Asked to update map output for unknown shuffle ${MDC(SHUFFLE_ID, shuffleId)}")
}
diff --git a/core/src/main/scala/org/apache/spark/SparkConf.scala b/core/src/main/scala/org/apache/spark/SparkConf.scala
index f1ef36bbf19c..fa7911c93737 100644
--- a/core/src/main/scala/org/apache/spark/SparkConf.scala
+++ b/core/src/main/scala/org/apache/spark/SparkConf.scala
@@ -25,7 +25,8 @@ import scala.jdk.CollectionConverters._
import org.apache.avro.{Schema, SchemaNormalization}
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.History._
import org.apache.spark.internal.config.Kryo._
@@ -507,11 +508,11 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging with Seria
// Used by Yarn in 1.1 and before
sys.props.get("spark.driver.libraryPath").foreach { value =>
val warning =
- s"""
- |spark.driver.libraryPath was detected (set to '$value').
+ log"""
+ |spark.driver.libraryPath was detected (set to '${MDC(LogKeys.CONFIG, value)}').
|This is deprecated in Spark 1.2+.
|
- |Please instead use: ${DRIVER_LIBRARY_PATH.key}
+ |Please instead use: ${MDC(LogKeys.CONFIG2, DRIVER_LIBRARY_PATH.key)}
""".stripMargin
logWarning(warning)
}
@@ -554,9 +555,13 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging with Seria
val executorCores = get(EXECUTOR_CORES)
val leftCores = totalCores % executorCores
if (leftCores != 0) {
- logWarning(s"Total executor cores: ${totalCores} is not " +
- s"divisible by cores per executor: ${executorCores}, " +
- s"the left cores: ${leftCores} will not be allocated")
+ logWarning(log"Total executor cores: " +
+ log"${MDC(LogKeys.NUM_EXECUTOR_CORES_TOTAL, totalCores)} " +
+ log"is not divisible by cores per executor: " +
+ log"${MDC(LogKeys.NUM_EXECUTOR_CORES, executorCores)}, " +
+ log"the left cores: " +
+ log"${MDC(LogKeys.NUM_EXECUTOR_CORES_REMAINING, leftCores)} " +
+ log"will not be allocated")
}
}
@@ -772,15 +777,20 @@ private[spark] object SparkConf extends Logging {
def logDeprecationWarning(key: String): Unit = {
deprecatedConfigs.get(key).foreach { cfg =>
logWarning(
- s"The configuration key '$key' has been deprecated as of Spark ${cfg.version} and " +
- s"may be removed in the future. ${cfg.deprecationMessage}")
+ log"The configuration key '${MDC(LogKeys.CONFIG, key)}' has been deprecated " +
+ log"as of Spark ${MDC(LogKeys.CONFIG_VERSION, cfg.version)} and " +
+ log"may be removed in the future. " +
+ log"${MDC(LogKeys.CONFIG_DEPRECATION_MESSAGE, cfg.deprecationMessage)}")
return
}
allAlternatives.get(key).foreach { case (newKey, cfg) =>
logWarning(
- s"The configuration key '$key' has been deprecated as of Spark ${cfg.version} and " +
- s"may be removed in the future. Please use the new key '$newKey' instead.")
+ log"The configuration key '${MDC(LogKeys.CONFIG, key)}' " +
+ log"has been deprecated as of " +
+ log"Spark ${MDC(LogKeys.CONFIG_VERSION, cfg.version)} and " +
+ log"may be removed in the future. Please use the new key " +
+ log"'${MDC(LogKeys.CONFIG_KEY_UPDATED, newKey)}' instead.")
return
}
}
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index b1c640dc935a..385f9cbc6cf4 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -747,8 +747,8 @@ class SparkContext(config: SparkConf) extends Logging {
case Some(endpointRef) =>
Some(endpointRef.askSync[Array[ThreadStackTrace]](TriggerThreadDump))
case None =>
- logWarning(s"Executor $executorId might already have stopped and " +
- "can not request thread dump from it.")
+ logWarning(log"Executor ${MDC(LogKeys.EXECUTOR_ID, executorId)} " +
+ log"might already have stopped and can not request thread dump from it.")
None
}
}
@@ -779,8 +779,8 @@ class SparkContext(config: SparkConf) extends Logging {
case Some(endpointRef) =>
Some(endpointRef.askSync[Array[String]](TriggerHeapHistogram))
case None =>
- logWarning(s"Executor $executorId might already have stopped and " +
- "can not request heap histogram from it.")
+ logWarning(log"Executor ${MDC(LogKeys.EXECUTOR_ID, executorId)} " +
+ log"might already have stopped and can not request heap histogram from it.")
None
}
}
@@ -1785,8 +1785,9 @@ class SparkContext(config: SparkConf) extends Logging {
val schemeCorrectedURI = uri.getScheme match {
case null => new File(path).getCanonicalFile.toURI
case "local" =>
- logWarning(s"File with 'local' scheme $path is not supported to add to file server, " +
- s"since it is already available on every node.")
+ logWarning(log"File with 'local' scheme ${MDC(LogKeys.PATH, path)} " +
+ log"is not supported to add to file server, " +
+ log"since it is already available on every node.")
return
case _ => uri
}
@@ -1859,8 +1860,9 @@ class SparkContext(config: SparkConf) extends Logging {
Utils.unpack(source, dest)
postEnvironmentUpdate()
} else {
- logWarning(s"The path $path has been added already. Overwriting of added paths " +
- "is not supported in the current version.")
+ logWarning(log"The path ${MDC(LogKeys.PATH, path)} " +
+ log"has been added already. Overwriting of added paths " +
+ log"is not supported in the current version.")
}
}
@@ -2220,8 +2222,9 @@ class SparkContext(config: SparkConf) extends Logging {
}
if (existed.nonEmpty) {
val jarMessage = if (scheme != "ivy") "JAR" else "dependency jars of Ivy URI"
- logWarning(s"The $jarMessage $path at ${existed.mkString(",")} has been added already." +
- " Overwriting of added jar is not supported in the current version.")
+ logWarning(log"The ${MDC(LogKeys.JAR_MESSAGE, jarMessage)} ${MDC(LogKeys.PATH, path)} " +
+ log"at ${MDC(LogKeys.EXISTING_PATH, existed.mkString(","))} has been added already." +
+ log" Overwriting of added jar is not supported in the current version.")
}
}
}
@@ -2742,9 +2745,9 @@ class SparkContext(config: SparkConf) extends Logging {
// its own local file system, which is incorrect because the checkpoint files
// are actually on the executor machines.
if (!isLocal && Utils.nonLocalPaths(directory).isEmpty) {
- logWarning("Spark is not running in local mode, therefore the checkpoint directory " +
- s"must not be on the local filesystem. Directory '$directory' " +
- "appears to be on the local filesystem.")
+ logWarning(log"Spark is not running in local mode, therefore the checkpoint directory " +
+ log"must not be on the local filesystem. Directory '${MDC(LogKeys.PATH, directory)}' " +
+ log"appears to be on the local filesystem.")
}
checkpointDir = Option(directory).map { dir =>
@@ -2905,10 +2908,11 @@ object SparkContext extends Logging {
// its creationSite field being null:
val otherContextCreationSite =
Option(otherContext.creationSite).map(_.longForm).getOrElse("unknown location")
- val warnMsg = "Another SparkContext is being constructed (or threw an exception in its" +
- " constructor). This may indicate an error, since only one SparkContext should be" +
- " running in this JVM (see SPARK-2243)." +
- s" The other SparkContext was created at:\n$otherContextCreationSite"
+ val warnMsg = log"Another SparkContext is being constructed (or threw an exception in its" +
+ log" constructor). This may indicate an error, since only one SparkContext should be" +
+ log" running in this JVM (see SPARK-2243)." +
+ log" The other SparkContext was created at:\n" +
+ log"${MDC(LogKeys.CONTEXT_CREATION_SITE, otherContextCreationSite)}"
logWarning(warnMsg)
}
}
diff --git a/core/src/main/scala/org/apache/spark/SparkEnv.scala b/core/src/main/scala/org/apache/spark/SparkEnv.scala
index e1c84d181a2f..728b24d51732 100644
--- a/core/src/main/scala/org/apache/spark/SparkEnv.scala
+++ b/core/src/main/scala/org/apache/spark/SparkEnv.scala
@@ -32,7 +32,8 @@ import org.apache.spark.annotation.DeveloperApi
import org.apache.spark.api.python.{PythonWorker, PythonWorkerFactory}
import org.apache.spark.broadcast.BroadcastManager
import org.apache.spark.executor.ExecutorBackend
-import org.apache.spark.internal.{config, Logging}
+import org.apache.spark.internal.{config, Logging, MDC}
+import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.config._
import org.apache.spark.memory.{MemoryManager, UnifiedMemoryManager}
import org.apache.spark.metrics.{MetricsSystem, MetricsSystemInstances}
@@ -130,7 +131,8 @@ class SparkEnv (
Utils.deleteRecursively(new File(path))
} catch {
case e: Exception =>
- logWarning(s"Exception while deleting Spark temp dir: $path", e)
+ logWarning(log"Exception while deleting Spark temp dir: " +
+ log"${MDC(LogKeys.PATH, path)}", e)
}
case None => // We just need to delete tmp dir created by driver, so do nothing on executor
}
diff --git a/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala b/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala
index 7ff782db210d..16f87146c708 100644
--- a/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala
+++ b/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala
@@ -31,7 +31,8 @@ import scala.util.control.NonFatal
import org.apache.spark._
import org.apache.spark.api.python.PythonFunction.PythonAccumulator
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.{BUFFER_SIZE, EXECUTOR_CORES, Python}
import org.apache.spark.internal.config.Python._
import org.apache.spark.rdd.InputFileBlockHolder
@@ -592,7 +593,8 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
// Mimic the task name used in `Executor` to help the user find out the task to blame.
val taskName = s"${context.partitionId()}.${context.attemptNumber()} " +
s"in stage ${context.stageId()} (TID ${context.taskAttemptId()})"
- logWarning(s"Incomplete task $taskName interrupted: Attempting to kill Python Worker")
+ logWarning(log"Incomplete task ${MDC(TASK_NAME, taskName)} " +
+ log"interrupted: Attempting to kill Python Worker")
env.destroyPythonWorker(
pythonExec, workerModule, daemonModule, envVars.asScala.toMap, worker)
} catch {
diff --git a/core/src/main/scala/org/apache/spark/api/python/PythonWorkerFactory.scala b/core/src/main/scala/org/apache/spark/api/python/PythonWorkerFactory.scala
index f8260e177cc3..3221a4900f6a 100644
--- a/core/src/main/scala/org/apache/spark/api/python/PythonWorkerFactory.scala
+++ b/core/src/main/scala/org/apache/spark/api/python/PythonWorkerFactory.scala
@@ -30,7 +30,8 @@ import scala.jdk.CollectionConverters._
import org.apache.spark._
import org.apache.spark.errors.SparkCoreErrors
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.security.SocketAuthHelper
import org.apache.spark.util.{RedirectThread, Utils}
@@ -107,7 +108,8 @@ private[spark] class PythonWorkerFactory(
case c: CancelledKeyException => /* pass */
}
}
- logWarning(s"Worker ${worker} process from idle queue is dead, discarding.")
+ logWarning(log"Worker ${MDC(WORKER, worker)} " +
+ log"process from idle queue is dead, discarding.")
stopWorker(worker)
}
}
diff --git a/core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala b/core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala
index 6a46b611019f..0fe57dd0bb0a 100644
--- a/core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala
+++ b/core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala
@@ -28,7 +28,8 @@ import net.razorvine.pickle.{Pickler, Unpickler}
import org.apache.spark.SparkException
import org.apache.spark.api.java.JavaRDD
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.rdd.RDD
import org.apache.spark.util.ArrayImplicits._
@@ -143,22 +144,26 @@ private[spark] object SerDeUtil extends Logging {
}
(kt, vt) match {
case (Failure(kf), Failure(vf)) =>
- logWarning(s"""
- |Failed to pickle Java object as key: ${t._1.getClass.getSimpleName}, falling back
- |to 'toString'. Error: ${kf.getMessage}""".stripMargin)
- logWarning(s"""
- |Failed to pickle Java object as value: ${t._2.getClass.getSimpleName}, falling back
- |to 'toString'. Error: ${vf.getMessage}""".stripMargin)
+ logWarning(log"""
+ |Failed to pickle Java object as key:
+ |${MDC(CLASS_NAME, t._1.getClass.getSimpleName)}, falling back
+ |to 'toString'. Error: ${MDC(ERROR, kf.getMessage)}""".stripMargin)
+ logWarning(log"""
+ |Failed to pickle Java object as value:
+ |${MDC(CLASS_NAME, t._2.getClass.getSimpleName)}, falling back
+ |to 'toString'. Error: ${MDC(ERROR, vf.getMessage)}""".stripMargin)
(true, true)
case (Failure(kf), _) =>
- logWarning(s"""
- |Failed to pickle Java object as key: ${t._1.getClass.getSimpleName}, falling back
- |to 'toString'. Error: ${kf.getMessage}""".stripMargin)
+ logWarning(log"""
+ |Failed to pickle Java object as key:
+ |${MDC(CLASS_NAME, t._1.getClass.getSimpleName)}, falling back
+ |to 'toString'. Error: ${MDC(ERROR, kf.getMessage)}""".stripMargin)
(true, false)
case (_, Failure(vf)) =>
- logWarning(s"""
- |Failed to pickle Java object as value: ${t._2.getClass.getSimpleName}, falling back
- |to 'toString'. Error: ${vf.getMessage}""".stripMargin)
+ logWarning(log"""
+ |Failed to pickle Java object as value:
+ |${MDC(CLASS_NAME, t._2.getClass.getSimpleName)}, falling back
+ |to 'toString'. Error: ${MDC(ERROR, vf.getMessage)}""".stripMargin)
(false, true)
case _ =>
(false, false)
diff --git a/core/src/main/scala/org/apache/spark/api/r/RBackendHandler.scala b/core/src/main/scala/org/apache/spark/api/r/RBackendHandler.scala
index 1922c22a7133..45ac90626468 100644
--- a/core/src/main/scala/org/apache/spark/api/r/RBackendHandler.scala
+++ b/core/src/main/scala/org/apache/spark/api/r/RBackendHandler.scala
@@ -27,7 +27,7 @@ import io.netty.handler.timeout.ReadTimeoutException
import org.apache.spark.{SparkConf, SparkEnv}
import org.apache.spark.api.r.SerDe._
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.{METHOD_NAME, OBJECT_ID}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.R._
import org.apache.spark.util.{ThreadUtils, Utils}
import org.apache.spark.util.ArrayImplicits._
@@ -155,10 +155,11 @@ private[r] class RBackendHandler(server: RBackend)
args)
if (index.isEmpty) {
- logWarning(s"cannot find matching method ${cls}.$methodName. "
- + s"Candidates are:")
+ logWarning(log"cannot find matching method " +
+ log"${MDC(CLASS_NAME, cls)}.${MDC(METHOD_NAME, methodName)}. Candidates are:")
selectedMethods.foreach { method =>
- logWarning(s"$methodName(${method.getParameterTypes.mkString(",")})")
+ logWarning(log"${MDC(METHOD_NAME, methodName)}(" +
+ log"${MDC(METHOD_PARAMETER_TYPES, method.getParameterTypes.mkString(","))})")
}
throw new Exception(s"No matched method found for $cls.$methodName")
}
@@ -176,10 +177,11 @@ private[r] class RBackendHandler(server: RBackend)
args)
if (index.isEmpty) {
- logWarning(s"cannot find matching constructor for ${cls}. "
- + s"Candidates are:")
+ logWarning(log"cannot find matching constructor for ${MDC(CLASS_NAME, cls)}. "
+ + log"Candidates are:")
ctors.foreach { ctor =>
- logWarning(s"$cls(${ctor.getParameterTypes.mkString(",")})")
+ logWarning(log"${MDC(CLASS_NAME, cls)}(" +
+ log"${MDC(METHOD_PARAMETER_TYPES, ctor.getParameterTypes.mkString(","))})")
}
throw new Exception(s"No matched constructor found for $cls")
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/Client.scala b/core/src/main/scala/org/apache/spark/deploy/Client.scala
index 4b0e12861f05..50ce6f82800d 100644
--- a/core/src/main/scala/org/apache/spark/deploy/Client.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/Client.scala
@@ -32,7 +32,7 @@ import org.apache.spark.deploy.DeployMessages._
import org.apache.spark.deploy.master.{DriverState, Master}
import org.apache.spark.deploy.master.DriverState.DriverState
import org.apache.spark.internal.{config, Logging, MDC}
-import org.apache.spark.internal.LogKeys.{DRIVER_ID, ERROR, HOST_PORT}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.Network.RPC_ASK_TIMEOUT
import org.apache.spark.resource.ResourceUtils
import org.apache.spark.rpc.{RpcAddress, RpcEndpointRef, RpcEnv, ThreadSafeRpcEndpoint}
@@ -135,7 +135,8 @@ private class ClientEndpoint(
masterEndpoint.ask[T](message).onComplete {
case Success(v) => self.send(v)
case Failure(e) =>
- logWarning(s"Error sending messages to master $masterEndpoint", e)
+ logWarning(log"Error sending messages to master " +
+ log"${MDC(MASTER_URL, masterEndpoint)}", e)
}(forwardMessageExecutionContext)
}
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/DriverTimeoutPlugin.scala b/core/src/main/scala/org/apache/spark/deploy/DriverTimeoutPlugin.scala
index 9b141d607572..736c23556ec1 100644
--- a/core/src/main/scala/org/apache/spark/deploy/DriverTimeoutPlugin.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/DriverTimeoutPlugin.scala
@@ -23,7 +23,8 @@ import scala.jdk.CollectionConverters._
import org.apache.spark.SparkContext
import org.apache.spark.api.plugin.{DriverPlugin, ExecutorPlugin, PluginContext, SparkPlugin}
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.DRIVER_TIMEOUT
import org.apache.spark.util.{SparkExitCode, ThreadUtils}
@@ -48,8 +49,9 @@ class DriverTimeoutDriverPlugin extends DriverPlugin with Logging {
logWarning("Disabled with the timeout value 0.")
} else {
val task: Runnable = () => {
- logWarning(s"Terminate Driver JVM because it runs after $timeout minute" +
- (if (timeout == 1) "" else "s"))
+ logWarning(log"Terminate Driver JVM because it runs after " +
+ log"${MDC(TIME_UNITS, timeout)} minute" +
+ (if (timeout == 1) log"" else log"s"))
// We cannot use 'SparkContext.stop' because SparkContext might be in abnormal situation.
System.exit(SparkExitCode.DRIVER_TIMEOUT)
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
index 77b73fcfac1d..9d93d91b7d2e 100644
--- a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
@@ -40,7 +40,7 @@ import org.apache.spark._
import org.apache.spark.api.r.RUtils
import org.apache.spark.deploy.rest._
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.CLASS_NAME
+import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.UI._
import org.apache.spark.launcher.SparkLauncher
@@ -196,8 +196,8 @@ private[spark] class SparkSubmit extends Logging {
} catch {
// Fail over to use the legacy submission gateway
case e: SubmitRestConnectionException =>
- logWarning(s"Master endpoint ${args.master} was not a REST server. " +
- "Falling back to legacy submission gateway instead.")
+ logWarning(log"Master endpoint ${MDC(LogKeys.MASTER_URL, args.master)} " +
+ log"was not a REST server. Falling back to legacy submission gateway instead.")
args.useRest = false
submit(args, false)
}
@@ -897,9 +897,11 @@ private[spark] class SparkSubmit extends Logging {
if (childClasspath.nonEmpty && isCustomClasspathInClusterModeDisallowed) {
childClasspath.clear()
- logWarning(s"Ignore classpath ${childClasspath.mkString(", ")} with proxy user specified " +
- s"in Cluster mode when ${ALLOW_CUSTOM_CLASSPATH_BY_PROXY_USER_IN_CLUSTER_MODE.key} is " +
- s"disabled")
+ logWarning(log"Ignore classpath " +
+ log"${MDC(LogKeys.CLASS_PATH, childClasspath.mkString(", "))} " +
+ log"with proxy user specified in Cluster mode when " +
+ log"${MDC(LogKeys.CONFIG, ALLOW_CUSTOM_CLASSPATH_BY_PROXY_USER_IN_CLUSTER_MODE.key)} is " +
+ log"disabled")
}
(childArgs.toSeq, childClasspath.toSeq, sparkConf, childMainClass)
@@ -970,7 +972,7 @@ private[spark] class SparkSubmit extends Logging {
mainClass = Utils.classForName(childMainClass)
} catch {
case e: ClassNotFoundException =>
- logError(log"Failed to load class ${MDC(CLASS_NAME, childMainClass)}.")
+ logError(log"Failed to load class ${MDC(LogKeys.CLASS_NAME, childMainClass)}.")
if (childMainClass.contains("thriftserver")) {
logInfo(s"Failed to load main class $childMainClass.")
logInfo("You need to build Spark with -Phive and -Phive-thriftserver.")
@@ -981,7 +983,7 @@ private[spark] class SparkSubmit extends Logging {
}
throw new SparkUserAppException(CLASS_NOT_FOUND_EXIT_STATUS)
case e: NoClassDefFoundError =>
- logError(log"Failed to load ${MDC(CLASS_NAME, childMainClass)}", e)
+ logError(log"Failed to load ${MDC(LogKeys.CLASS_NAME, childMainClass)}", e)
if (e.getMessage.contains("org/apache/hadoop/hive")) {
logInfo(s"Failed to load hive class.")
logInfo("You need to build Spark with -Phive and -Phive-thriftserver.")
diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala
index 534c14000614..bf0a07c47906 100644
--- a/core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala
@@ -28,7 +28,8 @@ import scala.util.Try
import org.apache.spark.{SparkConf, SparkException, SparkUserAppException}
import org.apache.spark.deploy.SparkSubmitAction._
-import org.apache.spark.internal.{config, Logging}
+import org.apache.spark.internal.{config, Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.DYN_ALLOCATION_ENABLED
import org.apache.spark.launcher.SparkSubmitArgumentsParser
import org.apache.spark.network.util.JavaUtils
@@ -142,7 +143,7 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
sparkProperties.keys.foreach { k =>
if (!k.startsWith("spark.")) {
sparkProperties -= k
- logWarning(s"Ignoring non-Spark config property: $k")
+ logWarning(log"Ignoring non-Spark config property: ${MDC(CONFIG, k)}")
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClient.scala b/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClient.scala
index b0ee6018970a..ec231610b857 100644
--- a/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClient.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClient.scala
@@ -29,7 +29,8 @@ import org.apache.spark.SparkConf
import org.apache.spark.deploy.{ApplicationDescription, ExecutorState}
import org.apache.spark.deploy.DeployMessages._
import org.apache.spark.deploy.master.Master
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.resource.ResourceProfile
import org.apache.spark.rpc._
import org.apache.spark.scheduler.ExecutorDecommissionInfo
@@ -109,7 +110,8 @@ private[spark] class StandaloneAppClient(
masterRef.send(RegisterApplication(appDescription, self))
} catch {
case ie: InterruptedException => // Cancelled
- case NonFatal(e) => logWarning(s"Failed to connect to master $masterAddress", e)
+ case NonFatal(e) => logWarning(log"Failed to connect to master " +
+ log"${MDC(MASTER_URL, masterAddress)}", e)
}
})
}
@@ -146,7 +148,8 @@ private[spark] class StandaloneAppClient(
private def sendToMaster(message: Any): Unit = {
master match {
case Some(masterRef) => masterRef.send(message)
- case None => logWarning(s"Drop $message because has not yet connected to master")
+ case None => logWarning(
+ log"Drop ${MDC(MESSAGE, message)} because has not yet connected to master")
}
}
@@ -237,14 +240,16 @@ private[spark] class StandaloneAppClient(
override def onDisconnected(address: RpcAddress): Unit = {
if (master.exists(_.address == address)) {
- logWarning(s"Connection to $address failed; waiting for master to reconnect...")
+ logWarning(
+ log"Connection to ${MDC(MASTER_URL, address)} failed; waiting for master to reconnect...")
markDisconnected()
}
}
override def onNetworkError(cause: Throwable, address: RpcAddress): Unit = {
if (isPossibleMaster(address)) {
- logWarning(s"Could not connect to $address: $cause")
+ logWarning(log"Could not connect to ${MDC(MASTER_URL, address)}: " +
+ log"${MDC(ERROR, cause)}")
}
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/ApplicationCache.scala b/core/src/main/scala/org/apache/spark/deploy/history/ApplicationCache.scala
index 662746cf0c78..21252d97b15c 100644
--- a/core/src/main/scala/org/apache/spark/deploy/history/ApplicationCache.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/history/ApplicationCache.scala
@@ -28,7 +28,8 @@ import jakarta.servlet.{DispatcherType, Filter, FilterChain, FilterConfig, Servl
import jakarta.servlet.http.{HttpServletRequest, HttpServletResponse}
import org.eclipse.jetty.servlet.FilterHolder
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.metrics.source.Source
import org.apache.spark.ui.SparkUI
import org.apache.spark.util.Clock
@@ -197,7 +198,7 @@ private[history] class ApplicationCache(
new CacheEntry(loadedUI, completed)
} catch {
case e: Exception =>
- logWarning(s"Failed to initialize application UI for $application", e)
+ logWarning(log"Failed to initialize application UI for ${MDC(APP_ID, application)}", e)
operations.detachSparkUI(appId, attemptId, loadedUI.ui)
throw e
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala b/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala
index 07a873ac704d..d66f528665fb 100644
--- a/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileCompactor.scala
@@ -28,7 +28,8 @@ import org.apache.hadoop.fs.{FileStatus, FileSystem, Path}
import org.apache.spark.SparkConf
import org.apache.spark.deploy.history.EventFilter.FilterStatistics
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys
import org.apache.spark.scheduler.ReplayListenerBus
import org.apache.spark.util.Utils
@@ -174,7 +175,7 @@ class EventLogFileCompactor(
case _: IOException =>
}
if (!deleted) {
- logWarning(s"Failed to remove ${file.getPath} / skip removing.")
+ logWarning(log"Failed to remove ${MDC(LogKeys.PATH, file.getPath)} / skip removing.")
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileWriters.scala b/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileWriters.scala
index e7eb05c85367..80f0078048c4 100644
--- a/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileWriters.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/history/EventLogFileWriters.scala
@@ -28,7 +28,9 @@ import org.apache.hadoop.fs.permission.FsPermission
import org.apache.spark.SparkConf
import org.apache.spark.deploy.SparkHadoopUtil
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.io.CompressionCodec
import org.apache.spark.util.Utils
@@ -82,7 +84,7 @@ abstract class EventLogFileWriter(
protected def initLogFile(path: Path)(fnSetupWriter: OutputStream => PrintWriter): Unit = {
if (shouldOverwrite && fileSystem.delete(path, true)) {
- logWarning(s"Event log $path already exists. Overwriting...")
+ logWarning(log"Event log ${MDC(LogKeys.PATH, path)} already exists. Overwriting...")
}
val defaultFs = FileSystem.getDefaultUri(hadoopConf).getScheme
@@ -131,9 +133,10 @@ abstract class EventLogFileWriter(
protected def renameFile(src: Path, dest: Path, overwrite: Boolean): Unit = {
if (fileSystem.exists(dest)) {
if (overwrite) {
- logWarning(s"Event log $dest already exists. Overwriting...")
+ logWarning(log"Event log ${MDC(EVENT_LOG_DESTINATION, dest)} already exists. " +
+ log"Overwriting...")
if (!fileSystem.delete(dest, true)) {
- logWarning(s"Error deleting $dest")
+ logWarning(log"Error deleting ${MDC(EVENT_LOG_DESTINATION, dest)}")
}
} else {
throw new IOException(s"Target log file already exists ($dest)")
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
index c1f5a279f2fb..53b239ddfd79 100644
--- a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
@@ -38,7 +38,8 @@ import org.apache.hadoop.security.AccessControlException
import org.apache.spark.{SecurityManager, SparkConf, SparkException}
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.PATH
+import org.apache.spark.internal.LogKeys
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.History._
import org.apache.spark.internal.config.Status._
@@ -569,12 +570,13 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
case _: FileNotFoundException => false
case _: NoSuchElementException => false
case NonFatal(e) =>
- logWarning(s"Error while reading new log ${reader.rootPath}", e)
+ logWarning(log"Error while reading new log " +
+ log"${MDC(PATH, reader.rootPath)}", e)
false
}
case NonFatal(e) =>
- logWarning(s"Error while filtering log ${reader.rootPath}", e)
+ logWarning(log"Error while filtering log ${MDC(PATH, reader.rootPath)}", e)
false
}
}
@@ -729,7 +731,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
throw e
case e: AccessControlException =>
// We don't have read permissions on the log file
- logWarning(s"Unable to read log $rootPath", e)
+ logWarning(log"Unable to read log ${MDC(PATH, rootPath)}", e)
markInaccessible(rootPath)
// SPARK-28157 We should remove this inaccessible entry from the KVStore
// to handle permission-only changes with the same file sizes later.
@@ -744,8 +746,9 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
// Do nothing, the application completed during processing, the final event log file
// will be processed by next around.
} else {
- logWarning(s"In-progress event log file does not exist: ${reader.rootPath}, " +
- s"neither does the final event log file: $finalFilePath.")
+ logWarning(log"In-progress event log file does not exist: " +
+ log"${MDC(PATH, reader.rootPath)}, " +
+ log"neither does the final event log file: ${MDC(FINAL_PATH, finalFilePath)}.")
}
case e: Exception =>
logError("Exception while merging application listings", e)
@@ -919,7 +922,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
case e: InterruptedException =>
throw e
case e: AccessControlException =>
- logWarning(s"Insufficient permission while compacting log for $rootPath", e)
+ logWarning(log"Insufficient permission while compacting log for ${MDC(PATH, rootPath)}", e)
case e: Exception =>
logError(log"Exception while compacting log for ${MDC(PATH, rootPath)}", e)
} finally {
@@ -1011,7 +1014,8 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
}
}
if (count > 0) {
- logWarning(s"Fail to clean up according to MAX_LOG_NUM policy ($maxNum).")
+ logWarning(log"Fail to clean up according to MAX_LOG_NUM policy " +
+ log"(${MDC(MAX_LOG_NUM_POLICY, maxNum)}).")
}
}
@@ -1275,8 +1279,9 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
case _: IOException if !retried =>
// compaction may touch the file(s) which app rebuild wants to read
// compaction wouldn't run in short interval, so try again...
- logWarning(s"Exception occurred while rebuilding log path ${attempt.logPath} - " +
- "trying again...")
+ logWarning(log"Exception occurred while rebuilding log path " +
+ log"${MDC(PATH, attempt.logPath)} - " +
+ log"trying again...")
store.close()
memoryManager.release(appId, attempt.info.attemptId)
retried = true
@@ -1303,7 +1308,8 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
memoryManager.release(appId, attempt.info.attemptId)
}
override def onSwitchToDiskStoreFail(e: Exception): Unit = {
- logWarning(s"Failed to switch to diskStore for app $appId / ${attempt.info.attemptId}", e)
+ logWarning(log"Failed to switch to diskStore for app ${MDC(APP_ID, appId)} / " +
+ log"${MDC(LogKeys.APP_ATTEMPT_ID, attempt.info.attemptId)}", e)
diskStore.close()
lease.rollback()
}
@@ -1343,7 +1349,8 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
case _: IOException if !retried =>
// compaction may touch the file(s) which app rebuild wants to read
// compaction wouldn't run in short interval, so try again...
- logWarning(s"Exception occurred while rebuilding app $appId - trying again...")
+ logWarning(log"Exception occurred while rebuilding app ${MDC(APP_ID, appId)} - " +
+ log"trying again...")
lease.rollback()
retried = true
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/HistoryServerDiskManager.scala b/core/src/main/scala/org/apache/spark/deploy/history/HistoryServerDiskManager.scala
index a84e1b181954..f7df0ec1ac48 100644
--- a/core/src/main/scala/org/apache/spark/deploy/history/HistoryServerDiskManager.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/history/HistoryServerDiskManager.scala
@@ -25,7 +25,8 @@ import scala.collection.mutable.{HashMap, ListBuffer}
import org.apache.commons.io.FileUtils
import org.apache.spark.SparkConf
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.History._
import org.apache.spark.internal.config.History.HybridStoreDiskBackend.ROCKSDB
import org.apache.spark.status.KVUtils
@@ -246,7 +247,8 @@ private class HistoryServerDiskManager(
logInfo(s"Deleted ${evicted.size} store(s) to free ${Utils.bytesToString(freed)} " +
s"(target = ${Utils.bytesToString(size)}).")
} else {
- logWarning(s"Unable to free any space to make room for ${Utils.bytesToString(size)}.")
+ logWarning(log"Unable to free any space to make room for " +
+ log"${MDC(NUM_BYTES, Utils.bytesToString(size))}.")
}
}
}
@@ -312,8 +314,9 @@ private class HistoryServerDiskManager(
if (committedUsage.get() > maxUsage) {
val current = Utils.bytesToString(committedUsage.get())
val max = Utils.bytesToString(maxUsage)
- logWarning(s"Commit of application $appId / $attemptId causes maximum disk usage to be " +
- s"exceeded ($current > $max)")
+ logWarning(log"Commit of application ${MDC(APP_ID, appId)} / " +
+ log"${MDC(APP_ATTEMPT_ID, attemptId)} causes maximum disk usage to be " +
+ log"exceeded (${MDC(NUM_BYTES, current)} > ${MDC(NUM_BYTES_MAX, max)}")
}
updateApplicationStoreInfo(appId, attemptId, newSize)
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/FileSystemPersistenceEngine.scala b/core/src/main/scala/org/apache/spark/deploy/master/FileSystemPersistenceEngine.scala
index fb067f10c5a4..4332544e4491 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/FileSystemPersistenceEngine.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/FileSystemPersistenceEngine.scala
@@ -22,7 +22,8 @@ import java.nio.file.{FileAlreadyExistsException, Files, Paths}
import scala.reflect.ClassTag
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.io.CompressionCodec
import org.apache.spark.serializer.{DeserializationStream, SerializationStream, Serializer}
import org.apache.spark.util.ArrayImplicits._
@@ -56,7 +57,7 @@ private[master] class FileSystemPersistenceEngine(
override def unpersist(name: String): Unit = {
val f = new File(dir + File.separator + name)
if (!f.delete()) {
- logWarning(s"Error deleting ${f.getPath()}")
+ logWarning(log"Error deleting ${MDC(PATH, f.getPath())}")
}
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
index 46be03fc82a3..74bc1c5b535f 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
@@ -33,7 +33,7 @@ import org.apache.spark.deploy.master.MasterMessages._
import org.apache.spark.deploy.master.ui.MasterWebUI
import org.apache.spark.deploy.rest.StandaloneRestServer
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.{APP_DESC, APP_ID, EXECUTOR_ID, RETRY_COUNT}
+import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.Deploy._
import org.apache.spark.internal.config.Deploy.WorkerSelectionPolicy._
@@ -314,12 +314,13 @@ private[deploy] class Master(
workerInfo.lastHeartbeat = System.currentTimeMillis()
case None =>
if (workers.map(_.id).contains(workerId)) {
- logWarning(s"Got heartbeat from unregistered worker $workerId." +
- " Asking it to re-register.")
+ logWarning(log"Got heartbeat from unregistered worker " +
+ log"${MDC(LogKeys.WORKER_ID, workerId)}. Asking it to re-register.")
worker.send(ReconnectWorker(masterUrl))
} else {
- logWarning(s"Got heartbeat from unregistered worker $workerId." +
- " This worker was never registered, so ignoring the heartbeat.")
+ logWarning(log"Got heartbeat from unregistered worker " +
+ log"${MDC(LogKeys.WORKER_ID, workerId)}. " +
+ log"This worker was never registered, so ignoring the heartbeat.")
}
}
@@ -329,7 +330,7 @@ private[deploy] class Master(
logInfo("Application has been re-registered: " + appId)
app.state = ApplicationState.WAITING
case None =>
- logWarning("Master change ack from unknown app: " + appId)
+ logWarning(log"Master change ack from unknown app: ${MDC(LogKeys.APP_ID, appId)}")
}
if (canCompleteRecovery) { completeRecovery() }
@@ -363,7 +364,7 @@ private[deploy] class Master(
}
}
case None =>
- logWarning("Scheduler state from unknown worker: " + workerId)
+ logWarning(log"Scheduler state from unknown worker: ${MDC(LogKeys.WORKER_ID, workerId)}")
}
if (canCompleteRecovery) { completeRecovery() }
@@ -389,7 +390,7 @@ private[deploy] class Master(
}
}
case None =>
- logWarning("Worker state from unknown worker: " + workerId)
+ logWarning(log"Worker state from unknown worker: ${MDC(LogKeys.WORKER_ID, workerId)}")
}
case UnregisterApplication(applicationId) =>
@@ -449,7 +450,8 @@ private[deploy] class Master(
context.reply(KillDriverResponse(self, driverId, success = true, msg))
case None =>
val msg = s"Driver $driverId has already finished or does not exist"
- logWarning(msg)
+ logWarning(log"Driver ${MDC(LogKeys.DRIVER_ID, driverId)} " +
+ log"has already finished or does not exist")
context.reply(KillDriverResponse(self, driverId, success = false, msg))
}
}
@@ -566,16 +568,17 @@ private[deploy] class Master(
&& maxExecutorRetries >= 0) { // < 0 disables this application-killing path
val execs = appInfo.executors.values
if (!execs.exists(_.state == ExecutorState.RUNNING)) {
- logError(log"Application ${MDC(APP_DESC, appInfo.desc.name)} " +
- log"with ID ${MDC(APP_ID, appInfo.id)} " +
- log"failed ${MDC(RETRY_COUNT, appInfo.retryCount)} times; removing it")
+ logError(log"Application ${MDC(LogKeys.APP_DESC, appInfo.desc.name)} " +
+ log"with ID ${MDC(LogKeys.APP_ID, appInfo.id)} " +
+ log"failed ${MDC(LogKeys.RETRY_COUNT, appInfo.retryCount)} times; removing it")
removeApplication(appInfo, ApplicationState.FAILED)
}
}
}
schedule()
case None =>
- logWarning(s"Got status update for unknown executor $appId/$execId")
+ logWarning(log"Got status update for unknown executor ${MDC(LogKeys.APP_ID, appId)}" +
+ log"/${MDC(LogKeys.EXECUTOR_ID, execId)}")
}
context.reply(true)
}
@@ -649,13 +652,15 @@ private[deploy] class Master(
// Reschedule drivers which were not claimed by any workers
drivers.filter(_.worker.isEmpty).foreach { d =>
- logWarning(s"Driver ${d.id} was not found after master recovery")
+ logWarning(log"Driver ${MDC(LogKeys.DRIVER_ID, d.id)} " +
+ log"was not found after master recovery")
if (d.desc.supervise) {
- logWarning(s"Re-launching ${d.id}")
+ logWarning(log"Re-launching ${MDC(LogKeys.DRIVER_ID, d.id)}")
relaunchDriver(d)
} else {
removeDriver(d.id, DriverState.ERROR, None)
- logWarning(s"Did not re-launch ${d.id} because it was not supervised")
+ logWarning(log"Did not re-launch " +
+ log"${MDC(LogKeys.DRIVER_ID, d.id)} because it was not supervised")
}
}
@@ -696,8 +701,8 @@ private[deploy] class Master(
schedule()
} else {
val workerAddress = worker.endpoint.address
- logWarning("Worker registration failed. Attempted to re-register worker at same " +
- "address: " + workerAddress)
+ logWarning(log"Worker registration failed. Attempted to re-register worker at same " +
+ log"address: ${MDC(LogKeys.WORKER_URL, workerAddress)}")
workerRef.send(RegisterWorkerFailed("Attempted to re-register worker at same address: "
+ workerAddress))
}
@@ -833,12 +838,13 @@ private[deploy] class Master(
case CORES_FREE_DESC => aliveWorkers.sortBy(w => (w.coresFree, w.id)).reverse
case MEMORY_FREE_ASC => aliveWorkers.sortBy(w => (w.memoryFree, w.id))
case MEMORY_FREE_DESC => aliveWorkers.sortBy(w => (w.memoryFree, w.id)).reverse
- case WORKER_ID => aliveWorkers.sortBy(_.id)
+ case WorkerSelectionPolicy.WORKER_ID => aliveWorkers.sortBy(_.id)
}
val appMayHang = waitingApps.length == 1 &&
waitingApps.head.executors.isEmpty && usableWorkers.isEmpty
if (appMayHang) {
- logWarning(s"App ${app.id} requires more resource than any of Workers could have.")
+ logWarning(log"App ${MDC(LogKeys.APP_ID, app.id)} requires more resource " +
+ log"than any of Workers could have.")
}
val assignedCores =
scheduleExecutorsOnWorkers(app, rpId, resourceDesc, usableWorkers, spreadOutApps)
@@ -953,7 +959,8 @@ private[deploy] class Master(
curPos = (curPos + 1) % numWorkersAlive
}
if (!launched && isClusterIdle) {
- logWarning(s"Driver ${driver.id} requires more resource than any of Workers could have.")
+ logWarning(log"Driver ${MDC(LogKeys.DRIVER_ID, driver.id)} " +
+ log"requires more resource than any of Workers could have.")
}
}
} else {
@@ -967,8 +974,8 @@ private[deploy] class Master(
launchDriver(worker, driver)
waitingDrivers -= driver
case _ =>
- logWarning(
- s"Driver ${driver.id} requires more resource than any of Workers could have.")
+ logWarning(log"Driver ${MDC(LogKeys.DRIVER_ID, driver.id)} " +
+ log"requires more resource than any of Workers could have.")
}
}
}
@@ -1057,8 +1064,9 @@ private[deploy] class Master(
// On recovery do not add a decommissioned executor
persistenceEngine.removeWorker(worker)
} else {
- logWarning("Skipping decommissioning worker %s on %s:%d as worker is already decommissioned".
- format(worker.id, worker.host, worker.port))
+ logWarning(log"Skipping decommissioning worker ${MDC(LogKeys.WORKER_ID, worker.id)} " +
+ log"on ${MDC(LogKeys.WORKER_HOST, worker.host)}:" +
+ log"${MDC(LogKeys.WORKER_PORT, worker.port)} as worker is already decommissioned")
}
}
@@ -1194,8 +1202,9 @@ private[deploy] class Master(
schedule()
true
case None =>
- logWarning(s"Unknown application $appId requested executors:" +
- s" ${resourceProfileToTotalExecs}.")
+ logWarning(log"Unknown application " +
+ log"${MDC(LogKeys.APP_ID, appId)} requested executors:" +
+ log" ${MDC(LogKeys.RESOURCE_PROFILE_TO_TOTAL_EXECS, resourceProfileToTotalExecs)}.")
false
}
}
@@ -1220,13 +1229,15 @@ private[deploy] class Master(
killExecutor(desc)
}
if (unknown.nonEmpty) {
- logWarning(s"Application $appId attempted to kill non-existent executors: "
- + unknown.mkString(", "))
+ logWarning(log"Application ${MDC(LogKeys.APP_ID, appId)} attempted to kill " +
+ log"non-existent executors: " +
+ log"${MDC(LogKeys.EXECUTOR_IDS, unknown.mkString(", "))}")
}
schedule()
true
case None =>
- logWarning(s"Unregistered application $appId requested us to kill executors!")
+ logWarning(log"Unregistered application ${MDC(LogKeys.APP_ID, appId)} " +
+ log"requested us to kill executors!")
false
}
}
@@ -1245,7 +1256,8 @@ private[deploy] class Master(
} catch {
case e: NumberFormatException =>
// scalastyle:off line.size.limit
- logError(log"Encountered executor with a non-integer ID: ${MDC(EXECUTOR_ID, executorId)}. Ignoring")
+ logError(log"Encountered executor with a non-integer ID: " +
+ log"${MDC(LogKeys.EXECUTOR_ID, executorId)}. Ignoring")
// scalastyle:on
None
}
@@ -1280,8 +1292,8 @@ private[deploy] class Master(
for (worker <- toRemove) {
if (worker.state != WorkerState.DEAD) {
val workerTimeoutSecs = TimeUnit.MILLISECONDS.toSeconds(workerTimeoutMs)
- logWarning("Removing %s because we got no heartbeat in %d seconds".format(
- worker.id, workerTimeoutSecs))
+ logWarning(log"Removing ${MDC(LogKeys.WORKER_ID, worker.id)} because we got no heartbeat " +
+ log"in ${MDC(LogKeys.TIME_UNITS, workerTimeoutMs)} ms")
removeWorker(worker, s"Not receiving heartbeat for $workerTimeoutSecs seconds")
} else {
if (worker.lastHeartbeat < currentTime - ((reaperIterations + 1) * workerTimeoutMs)) {
@@ -1331,7 +1343,7 @@ private[deploy] class Master(
driver.worker.foreach(w => w.removeDriver(driver))
schedule()
case None =>
- logWarning(s"Asked to remove unknown driver: $driverId")
+ logWarning(log"Asked to remove unknown driver: ${MDC(LogKeys.DRIVER_ID, driverId)}")
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala b/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala
index a6d128de164f..247504f5ebbb 100644
--- a/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/rest/RestSubmissionClient.scala
@@ -34,7 +34,7 @@ import jakarta.servlet.http.HttpServletResponse
import org.apache.spark.{SPARK_VERSION => sparkVersion, SparkConf, SparkException}
import org.apache.spark.deploy.SparkApplication
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.{CLASS_NAME, ERROR, SUBMISSION_ID}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.util.Utils
/**
@@ -509,7 +509,7 @@ private[spark] class RestSubmissionClient(master: String) extends Logging {
*/
private def handleConnectionException(masterUrl: String): Boolean = {
if (!lostMasters.contains(masterUrl)) {
- logWarning(s"Unable to connect to server ${masterUrl}.")
+ logWarning(log"Unable to connect to server ${MDC(MASTER_URL, masterUrl)}.")
lostMasters += masterUrl
}
lostMasters.size >= masters.length
diff --git a/core/src/main/scala/org/apache/spark/deploy/security/HBaseDelegationTokenProvider.scala b/core/src/main/scala/org/apache/spark/deploy/security/HBaseDelegationTokenProvider.scala
index 1d13deed2af5..68df753e6b67 100644
--- a/core/src/main/scala/org/apache/spark/deploy/security/HBaseDelegationTokenProvider.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/security/HBaseDelegationTokenProvider.scala
@@ -28,7 +28,7 @@ import org.apache.hadoop.security.token.{Token, TokenIdentifier}
import org.apache.spark.SparkConf
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.SERVICE_NAME
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.security.HadoopDelegationTokenProvider
import org.apache.spark.util.Utils
diff --git a/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala b/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala
index 54a24927ded4..a079e0a19f46 100644
--- a/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala
@@ -31,7 +31,8 @@ import org.apache.hadoop.security.{Credentials, UserGroupInformation}
import org.apache.spark.SparkConf
import org.apache.spark.deploy.SparkHadoopUtil
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.config._
import org.apache.spark.rpc.RpcEndpointRef
import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.UpdateDelegationTokens
@@ -211,8 +212,9 @@ private[spark] class HadoopDelegationTokenManager(
null
case e: Exception =>
val delay = TimeUnit.SECONDS.toMillis(sparkConf.get(CREDENTIALS_RENEWAL_RETRY_WAIT))
- logWarning(s"Failed to update tokens, will try again in ${UIUtils.formatDuration(delay)}!" +
- " If this happens too often tasks will fail.", e)
+ logWarning(log"Failed to update tokens, will try again in " +
+ log"${MDC(LogKeys.TIME_UNITS, UIUtils.formatDuration(delay))}!" +
+ log" If this happens too often tasks will fail.", e)
scheduleRenewal(delay)
null
}
@@ -296,7 +298,8 @@ private[spark] object HadoopDelegationTokenManager extends Logging {
deprecatedProviderEnabledConfigs.foreach { pattern =>
val deprecatedKey = pattern.format(serviceName)
if (sparkConf.contains(deprecatedKey)) {
- logWarning(s"${deprecatedKey} is deprecated. Please use ${key} instead.")
+ logWarning(log"${MDC(LogKeys.DEPRECATED_KEY, deprecatedKey)} is deprecated. " +
+ log"Please use ${MDC(LogKeys.CONFIG, key)} instead.")
}
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/security/HadoopFSDelegationTokenProvider.scala b/core/src/main/scala/org/apache/spark/deploy/security/HadoopFSDelegationTokenProvider.scala
index 8eb45238b477..bc1c31b08aff 100644
--- a/core/src/main/scala/org/apache/spark/deploy/security/HadoopFSDelegationTokenProvider.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/security/HadoopFSDelegationTokenProvider.scala
@@ -28,7 +28,8 @@ import org.apache.hadoop.security.{Credentials, UserGroupInformation}
import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenIdentifier
import org.apache.spark.{SparkConf, SparkException}
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.security.HadoopDelegationTokenProvider
import org.apache.spark.util.Utils
@@ -78,7 +79,7 @@ private[deploy] class HadoopFSDelegationTokenProvider
nextRenewalDate
} catch {
case NonFatal(e) =>
- logWarning(s"Failed to get token from service $serviceName", e)
+ logWarning(log"Failed to get token from service ${MDC(SERVICE_NAME, serviceName)}", e)
None
}
}
@@ -157,17 +158,20 @@ private[deploy] class HadoopFSDelegationTokenProvider
val now = System.currentTimeMillis()
val issueDate = identifier.getIssueDate
if (issueDate > now) {
- logWarning(s"Token $kind has set up issue date later than current time. (provided: " +
- s"$issueDate / current timestamp: $now) Please make sure clocks are in sync between " +
- "machines. If the issue is not a clock mismatch, consult token implementor to check " +
- "whether issue date is valid.")
+ logWarning(log"Token ${MDC(TOKEN_KIND, kind)} has set up issue date later than " +
+ log"current time (provided: " +
+ log"${MDC(ISSUE_DATE, issueDate)} / current timestamp: ${MDC(CURRENT_TIME, now)}). " +
+ log"Please make sure clocks are in sync between " +
+ log"machines. If the issue is not a clock mismatch, consult token implementor to check " +
+ log"whether issue date is valid.")
issueDate
} else if (issueDate > 0L) {
issueDate
} else {
- logWarning(s"Token $kind has not set up issue date properly. (provided: $issueDate) " +
- s"Using current timestamp ($now) as issue date instead. Consult token implementor to fix " +
- "the behavior.")
+ logWarning(log"Token ${MDC(TOKEN_KIND, kind)} has not set up issue date properly " +
+ log"(provided: ${MDC(ISSUE_DATE, issueDate)}). " +
+ log"Using current timestamp (${MDC(CURRENT_TIME, now)} as issue date instead. " +
+ log"Consult token implementor to fix the behavior.")
now
}
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/DriverRunner.scala b/core/src/main/scala/org/apache/spark/deploy/worker/DriverRunner.scala
index e7fca402a887..8ab1e57ce77d 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/DriverRunner.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/DriverRunner.scala
@@ -31,7 +31,8 @@ import org.apache.spark.deploy.DeployMessages.DriverStateChanged
import org.apache.spark.deploy.StandaloneResourceUtils.prepareResourcesFile
import org.apache.spark.deploy.master.DriverState
import org.apache.spark.deploy.master.DriverState.DriverState
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.{DRIVER_RESOURCES_FILE, SPARK_DRIVER_PREFIX}
import org.apache.spark.internal.config.UI.UI_REVERSE_PROXY
import org.apache.spark.internal.config.Worker.WORKER_DRIVER_TERMINATE_TIMEOUT
@@ -131,8 +132,8 @@ private[deploy] class DriverRunner(
process.foreach { p =>
val exitCode = Utils.terminateProcess(p, driverTerminateTimeoutMs)
if (exitCode.isEmpty) {
- logWarning("Failed to terminate driver process: " + p +
- ". This process will likely be orphaned.")
+ logWarning(log"Failed to terminate driver process: ${MDC(PROCESS, p)} " +
+ log". This process will likely be orphaned.")
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/ExecutorRunner.scala b/core/src/main/scala/org/apache/spark/deploy/worker/ExecutorRunner.scala
index 5547593a28f5..39a599a6532c 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/ExecutorRunner.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/ExecutorRunner.scala
@@ -28,7 +28,8 @@ import org.apache.spark.{SecurityManager, SparkConf}
import org.apache.spark.deploy.{ApplicationDescription, ExecutorState}
import org.apache.spark.deploy.DeployMessages.ExecutorStateChanged
import org.apache.spark.deploy.StandaloneResourceUtils.prepareResourcesFile
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.SPARK_EXECUTOR_PREFIX
import org.apache.spark.internal.config.UI._
import org.apache.spark.resource.ResourceInformation
@@ -107,14 +108,14 @@ private[deploy] class ExecutorRunner(
}
exitCode = Utils.terminateProcess(process, EXECUTOR_TERMINATE_TIMEOUT_MS)
if (exitCode.isEmpty) {
- logWarning("Failed to terminate process: " + process +
- ". This process will likely be orphaned.")
+ logWarning(log"Failed to terminate process: ${MDC(PROCESS, process)}" +
+ log". This process will likely be orphaned.")
}
}
try {
worker.send(ExecutorStateChanged(appId, execId, state, message, exitCode))
} catch {
- case e: IllegalStateException => logWarning(e.getMessage(), e)
+ case e: IllegalStateException => logWarning(log"${MDC(ERROR, e.getMessage())}", e)
}
}
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala b/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
index 302da3473809..f030475131d2 100755
--- a/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
@@ -38,6 +38,7 @@ import org.apache.spark.deploy.StandaloneResourceUtils._
import org.apache.spark.deploy.master.{DriverState, Master}
import org.apache.spark.deploy.worker.ui.WorkerWebUI
import org.apache.spark.internal.{config, Logging, MDC}
+import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.Tests.IS_TESTING
import org.apache.spark.internal.config.UI._
@@ -74,8 +75,8 @@ private[deploy] class Worker(
if (conf.get(config.DECOMMISSION_ENABLED)) {
val signal = conf.get(config.Worker.WORKER_DECOMMISSION_SIGNAL)
logInfo(s"Registering SIG$signal handler to trigger decommissioning.")
- SignalUtils.register(signal, s"Failed to register SIG$signal handler - " +
- "disabling worker decommission feature.") {
+ SignalUtils.register(signal, log"Failed to register SIG${MDC(LogKeys.SIGNAL, signal)} " +
+ log"handler - disabling worker decommission feature.") {
self.send(WorkerDecommissionSigReceived)
true
}
@@ -322,7 +323,8 @@ private[deploy] class Worker(
sendRegisterMessageToMaster(masterEndpoint)
} catch {
case ie: InterruptedException => // Cancelled
- case NonFatal(e) => logWarning(s"Failed to connect to master $masterAddress", e)
+ case NonFatal(e) => logWarning(
+ log"Failed to connect to master ${MDC(MASTER_URL, masterAddress)}", e)
}
}
})
@@ -379,7 +381,9 @@ private[deploy] class Worker(
sendRegisterMessageToMaster(masterEndpoint)
} catch {
case ie: InterruptedException => // Cancelled
- case NonFatal(e) => logWarning(s"Failed to connect to master $masterAddress", e)
+ case NonFatal(e) =>
+ logWarning(log"Failed to connect to master " +
+ log"${MDC(MASTER_URL, masterAddress)}", e)
}
}
}))
@@ -475,7 +479,8 @@ private[deploy] class Worker(
// e.g. Master disconnect(maybe due to network drop) and recover immediately, see
// SPARK-23191 for more details.
if (duplicate) {
- logWarning(s"Duplicate registration at master $preferredMasterAddress")
+ logWarning(log"Duplicate registration at master " +
+ log"${MDC(MASTER_URL, preferredMasterAddress)}")
}
logInfo(s"Successfully registered with master $preferredMasterAddress")
@@ -575,7 +580,8 @@ private[deploy] class Worker(
case LaunchExecutor(masterUrl, appId, execId, rpId, appDesc, cores_, memory_, resources_) =>
if (masterUrl != activeMasterUrl) {
- logWarning("Invalid Master (" + masterUrl + ") attempted to launch executor.")
+ logWarning(log"Invalid Master (${MDC(MASTER_URL, masterUrl)}) " +
+ log"attempted to launch executor.")
} else if (decommissioned) {
logWarning("Asked to launch an executor while decommissioned. Not launching executor.")
} else {
@@ -600,7 +606,7 @@ private[deploy] class Worker(
Some(appDir.getAbsolutePath())
} catch {
case e: IOException =>
- logWarning(s"${e.getMessage}. Ignoring this directory.")
+ logWarning(log"${MDC(ERROR, e.getMessage)}. Ignoring this directory.")
None
}
}.toImmutableArraySeq
@@ -655,7 +661,8 @@ private[deploy] class Worker(
case KillExecutor(masterUrl, appId, execId) =>
if (masterUrl != activeMasterUrl) {
- logWarning("Invalid Master (" + masterUrl + ") attempted to kill executor " + execId)
+ logWarning(log"Invalid Master (${MDC(MASTER_URL, masterUrl)}) " +
+ log"attempted to kill executor ${MDC(EXECUTOR_ID, execId)}")
} else {
val fullId = appId + "/" + execId
executors.get(fullId) match {
@@ -771,7 +778,8 @@ private[deploy] class Worker(
case Some(masterRef) => masterRef.send(message)
case None =>
logWarning(
- s"Dropping $message because the connection to master has not yet been established")
+ log"Dropping ${MDC(MESSAGE, message)} " +
+ log"because the connection to master has not yet been established")
}
}
@@ -821,7 +829,8 @@ private[deploy] class Worker(
case None =>
logWarning(
- s"Dropping $newState because the connection to master has not yet been established")
+ log"Dropping ${MDC(NEW_STATE, newState)} " +
+ log"because the connection to master has not yet been established")
}
}
@@ -867,9 +876,9 @@ private[deploy] class Worker(
decommissioned = true
logInfo(s"Decommission worker $workerId.")
} else if (decommissioned) {
- logWarning(s"Worker $workerId already started decommissioning.")
+ logWarning(log"Worker ${MDC(WORKER_ID, workerId)} already started decommissioning.")
} else {
- logWarning(s"Receive decommission request, but decommission feature is disabled.")
+ logWarning("Receive decommission request, but decommission feature is disabled.")
}
}
@@ -879,13 +888,15 @@ private[deploy] class Worker(
val state = driverStateChanged.state
state match {
case DriverState.ERROR =>
- logWarning(s"Driver $driverId failed with unrecoverable exception: ${exception.get}")
+ logWarning(log"Driver ${MDC(DRIVER_ID, driverId)} " +
+ log"failed with unrecoverable exception: ${MDC(ERROR, exception.get)}")
case DriverState.FAILED =>
- logWarning(s"Driver $driverId exited with failure")
+ logWarning(log"Driver ${MDC(DRIVER_ID, driverId)} exited with failure")
case DriverState.FINISHED =>
registrationRetryTimer match {
case Some(_) =>
- logWarning(s"Driver $driverId exited successfully while master is disconnected.")
+ logWarning(log"Driver ${MDC(DRIVER_ID, driverId)} " +
+ log"exited successfully while master is disconnected.")
case _ =>
logInfo(s"Driver $driverId exited successfully")
}
diff --git a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
index bead4d8dd2a5..21b30cf854e6 100644
--- a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
+++ b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
@@ -32,7 +32,7 @@ import org.apache.spark.TaskState.TaskState
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.deploy.worker.WorkerWatcher
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.{CONFIG, CONFIG2, REASON}
+import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.config._
import org.apache.spark.network.netty.SparkTransportConf
import org.apache.spark.network.util.NettyUtils
@@ -75,8 +75,9 @@ private[spark] class CoarseGrainedExecutorBackend(
if (env.conf.get(DECOMMISSION_ENABLED)) {
val signal = env.conf.get(EXECUTOR_DECOMMISSION_SIGNAL)
logInfo(s"Registering SIG$signal handler to trigger decommissioning.")
- SignalUtils.register(signal, s"Failed to register SIG$signal handler - disabling" +
- s" executor decommission feature.") (self.askSync[Boolean](ExecutorDecommissionSigReceived))
+ SignalUtils.register(signal, log"Failed to register SIG${MDC(LogKeys.SIGNAL, signal)} " +
+ log"handler - disabling executor decommission feature.")(
+ self.askSync[Boolean](ExecutorDecommissionSigReceived))
}
logInfo("Connecting to driver: " + driverUrl)
@@ -256,7 +257,8 @@ private[spark] class CoarseGrainedExecutorBackend(
exitExecutor(1, s"Driver $remoteAddress disassociated! Shutting down.", null,
notifyDriver = false)
} else {
- logWarning(s"An unknown ($remoteAddress) driver disconnected.")
+ logWarning(log"An unknown (${MDC(LogKeys.REMOTE_ADDRESS, remoteAddress)} " +
+ log"driver disconnected.")
}
}
@@ -269,7 +271,8 @@ private[spark] class CoarseGrainedExecutorBackend(
}
driver match {
case Some(driverRef) => driverRef.send(msg)
- case None => logWarning(s"Drop $msg because has not yet connected to driver")
+ case None =>
+ logWarning(log"Drop ${MDC(LogKeys.MESSAGE, msg)} because has not yet connected to driver")
}
}
@@ -283,7 +286,7 @@ private[spark] class CoarseGrainedExecutorBackend(
throwable: Throwable = null,
notifyDriver: Boolean = true) = {
if (stopping.compareAndSet(false, true)) {
- val message = log"Executor self-exiting due to : ${MDC(REASON, reason)}"
+ val message = log"Executor self-exiting due to : ${MDC(LogKeys.REASON, reason)}"
if (throwable != null) {
logError(message, throwable)
} else {
@@ -305,10 +308,11 @@ private[spark] class CoarseGrainedExecutorBackend(
private def decommissionSelf(): Unit = {
if (!env.conf.get(DECOMMISSION_ENABLED)) {
- logWarning(s"Receive decommission request, but decommission feature is disabled.")
+ logWarning("Receive decommission request, but decommission feature is disabled.")
return
} else if (decommissioned) {
- logWarning(s"Executor $executorId already started decommissioning.")
+ logWarning(log"Executor ${MDC(LogKeys.EXECUTOR_ID, executorId)} " +
+ log"already started decommissioning.")
return
}
val msg = s"Decommission executor $executorId."
@@ -322,8 +326,8 @@ private[spark] class CoarseGrainedExecutorBackend(
env.blockManager.decommissionBlockManager()
} else if (env.conf.get(STORAGE_DECOMMISSION_ENABLED)) {
logError(log"Storage decommissioning attempted but neither " +
- log"${MDC(CONFIG, STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED.key)} or " +
- log"${MDC(CONFIG2, STORAGE_DECOMMISSION_RDD_BLOCKS_ENABLED.key)} is enabled ")
+ log"${MDC(LogKeys.CONFIG, STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED.key)} or " +
+ log"${MDC(LogKeys.CONFIG2, STORAGE_DECOMMISSION_RDD_BLOCKS_ENABLED.key)} is enabled ")
}
if (executor != null) {
executor.decommission()
diff --git a/core/src/main/scala/org/apache/spark/executor/Executor.scala b/core/src/main/scala/org/apache/spark/executor/Executor.scala
index ea24dd57b43f..fd6c02c07789 100644
--- a/core/src/main/scala/org/apache/spark/executor/Executor.scala
+++ b/core/src/main/scala/org/apache/spark/executor/Executor.scala
@@ -41,7 +41,7 @@ import org.slf4j.MDC
import org.apache.spark._
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.internal.{Logging, MDC => LogMDC}
-import org.apache.spark.internal.LogKeys.{CLASS_NAME, ERROR, MAX_ATTEMPTS, TASK_ID, TASK_NAME, TIMEOUT}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.internal.plugin.PluginContainer
import org.apache.spark.memory.{SparkOutOfMemoryError, TaskMemoryManager}
@@ -638,9 +638,10 @@ private[spark] class Executor(
val freedMemory = taskMemoryManager.cleanUpAllAllocatedMemory()
if (freedMemory > 0 && !threwException) {
- val errMsg = s"Managed memory leak detected; size = $freedMemory bytes, $taskName"
+ val errMsg = log"Managed memory leak detected; size = " +
+ log"${LogMDC(NUM_BYTES, freedMemory)} bytes, ${LogMDC(TASK_NAME, taskName)}"
if (conf.get(UNSAFE_EXCEPTION_ON_MEMORY_LEAK)) {
- throw SparkException.internalError(errMsg, category = "EXECUTOR")
+ throw SparkException.internalError(errMsg.message, category = "EXECUTOR")
} else {
logWarning(errMsg)
}
@@ -727,9 +728,11 @@ private[spark] class Executor(
// directSend = sending directly back to the driver
val serializedResult: ByteBuffer = {
if (maxResultSize > 0 && resultSize > maxResultSize) {
- logWarning(s"Finished $taskName. Result is larger than maxResultSize " +
- s"(${Utils.bytesToString(resultSize)} > ${Utils.bytesToString(maxResultSize)}), " +
- s"dropping it.")
+ logWarning(log"Finished ${LogMDC(TASK_NAME, taskName)}. " +
+ log"Result is larger than maxResultSize " +
+ log"(${LogMDC(RESULT_SIZE_BYTES, Utils.bytesToString(resultSize))} > " +
+ log"${LogMDC(RESULT_SIZE_BYTES_MAX, Utils.bytesToString(maxResultSize))}), " +
+ log"dropping it.")
ser.serialize(new IndirectTaskResult[Any](TaskResultBlockId(taskId), resultSize))
} else if (resultSize > maxDirectResultSize) {
val blockId = TaskResultBlockId(taskId)
@@ -778,11 +781,12 @@ private[spark] class Executor(
if (!t.isInstanceOf[FetchFailedException]) {
// there was a fetch failure in the task, but some user code wrapped that exception
// and threw something else. Regardless, we treat it as a fetch failure.
- val fetchFailedCls = classOf[FetchFailedException].getName
- logWarning(s"$taskName encountered a ${fetchFailedCls} and " +
- s"failed, but the ${fetchFailedCls} was hidden by another " +
- s"exception. Spark is handling this like a fetch failure and ignoring the " +
- s"other exception: $t")
+ logWarning(log"${LogMDC(TASK_NAME, taskName)} encountered a " +
+ log"${LogMDC(CLASS_NAME, classOf[FetchFailedException].getName)} " +
+ log"and failed, but the " +
+ log"${LogMDC(CLASS_NAME, classOf[FetchFailedException].getName)} " +
+ log"was hidden by another exception. Spark is handling this like a fetch failure " +
+ log"and ignoring the other exception: ${LogMDC(ERROR, t)}")
}
setTaskFinishedAndClearInterruptStatus()
plugins.foreach(_.onTaskFailed(reason))
@@ -994,12 +998,14 @@ private[spark] class Executor(
finished = true
} else {
val elapsedTimeMs = TimeUnit.NANOSECONDS.toMillis(elapsedTimeNs)
- logWarning(s"Killed task $taskId is still running after $elapsedTimeMs ms")
+ logWarning(log"Killed task ${LogMDC(TASK_ID, taskId)} " +
+ log"is still running after ${LogMDC(TIME_UNITS, elapsedTimeMs)} ms")
if (takeThreadDump) {
try {
taskRunner.theadDump().foreach { thread =>
if (thread.threadName == taskRunner.threadName) {
- logWarning(s"Thread dump from task $taskId:\n${thread.toString}")
+ logWarning(log"Thread dump from task ${LogMDC(TASK_ID, taskId)}:\n" +
+ log"${LogMDC(THREAD, thread.toString)}")
}
}
} catch {
@@ -1259,7 +1265,7 @@ private[spark] class Executor(
if (runner != null) {
runner.theadDump()
} else {
- logWarning(s"Failed to dump thread for task $taskId")
+ logWarning(log"Failed to dump thread for task ${LogMDC(TASK_ID, taskId)}")
None
}
}
diff --git a/core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala b/core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
index 3a24da98ecc2..f245d2d4e407 100644
--- a/core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
+++ b/core/src/main/scala/org/apache/spark/internal/io/HadoopMapReduceCommitProtocol.scala
@@ -29,7 +29,8 @@ import org.apache.hadoop.mapreduce._
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.mapred.SparkHadoopMapRedUtil
/**
@@ -252,7 +253,7 @@ class HadoopMapReduceCommitProtocol(
committer.abortJob(jobContext, JobStatus.State.FAILED)
} catch {
case e: IOException =>
- logWarning(s"Exception while aborting ${jobContext.getJobID}", e)
+ logWarning(log"Exception while aborting ${MDC(JOB_ID, jobContext.getJobID)}", e)
}
try {
if (hasValidPath) {
@@ -261,7 +262,7 @@ class HadoopMapReduceCommitProtocol(
}
} catch {
case e: IOException =>
- logWarning(s"Exception while aborting ${jobContext.getJobID}", e)
+ logWarning(log"Exception while aborting ${MDC(JOB_ID, jobContext.getJobID)}", e)
}
}
@@ -292,7 +293,8 @@ class HadoopMapReduceCommitProtocol(
committer.abortTask(taskContext)
} catch {
case e: IOException =>
- logWarning(s"Exception while aborting ${taskContext.getTaskAttemptID}", e)
+ logWarning(log"Exception while aborting " +
+ log"${MDC(TASK_ATTEMPT_ID, taskContext.getTaskAttemptID)}", e)
}
// best effort cleanup of other staged files
try {
@@ -302,7 +304,8 @@ class HadoopMapReduceCommitProtocol(
}
} catch {
case e: IOException =>
- logWarning(s"Exception while aborting ${taskContext.getTaskAttemptID}", e)
+ logWarning(log"Exception while aborting " +
+ log"${MDC(TASK_ATTEMPT_ID, taskContext.getTaskAttemptID)}", e)
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/internal/plugin/PluginEndpoint.scala b/core/src/main/scala/org/apache/spark/internal/plugin/PluginEndpoint.scala
index bc45aefa560e..6ff918979c9e 100644
--- a/core/src/main/scala/org/apache/spark/internal/plugin/PluginEndpoint.scala
+++ b/core/src/main/scala/org/apache/spark/internal/plugin/PluginEndpoint.scala
@@ -18,7 +18,8 @@
package org.apache.spark.internal.plugin
import org.apache.spark.api.plugin.DriverPlugin
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.rpc.{IsolatedThreadSafeRpcEndpoint, RpcCallContext, RpcEnv}
case class PluginMessage(pluginName: String, message: AnyRef)
@@ -36,13 +37,15 @@ private class PluginEndpoint(
val reply = plugin.receive(message)
if (reply != null) {
logWarning(
- s"Plugin $pluginName returned reply for one-way message of type " +
- s"${message.getClass().getName()}.")
+ log"Plugin ${MDC(PLUGIN_NAME, pluginName)} " +
+ log"returned reply for one-way message of type " +
+ log"${MDC(CLASS_NAME, message.getClass().getName())}.")
}
} catch {
case e: Exception =>
- logWarning(s"Error in plugin $pluginName when handling message of type " +
- s"${message.getClass().getName()}.", e)
+ logWarning(log"Error in plugin ${MDC(PLUGIN_NAME, pluginName)} " +
+ log"when handling message of type " +
+ log"${MDC(CLASS_NAME, message.getClass().getName())}.", e)
}
case None =>
diff --git a/core/src/main/scala/org/apache/spark/memory/ExecutionMemoryPool.scala b/core/src/main/scala/org/apache/spark/memory/ExecutionMemoryPool.scala
index 4cffbb2a5701..98846e064756 100644
--- a/core/src/main/scala/org/apache/spark/memory/ExecutionMemoryPool.scala
+++ b/core/src/main/scala/org/apache/spark/memory/ExecutionMemoryPool.scala
@@ -21,7 +21,8 @@ import javax.annotation.concurrent.GuardedBy
import scala.collection.mutable
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
/**
* Implements policies and bookkeeping for sharing an adjustable-sized pool of memory between tasks.
@@ -153,8 +154,9 @@ private[memory] class ExecutionMemoryPool(
val curMem = memoryForTask.getOrElse(taskAttemptId, 0L)
val memoryToFree = if (curMem < numBytes) {
logWarning(
- s"Internal error: release called on $numBytes bytes but task only has $curMem bytes " +
- s"of memory from the $poolName pool")
+ log"Internal error: release called on ${MDC(NUM_BYTES, numBytes)} " +
+ log"bytes but task only has ${MDC(CURRENT_MEMORY_BYTES, curMem)} bytes " +
+ log"of memory from the ${MDC(MEMORY_POOL_NAME, poolName)} pool")
curMem
} else {
numBytes
diff --git a/core/src/main/scala/org/apache/spark/memory/StorageMemoryPool.scala b/core/src/main/scala/org/apache/spark/memory/StorageMemoryPool.scala
index 0abdca99de1b..24fcb5b17f38 100644
--- a/core/src/main/scala/org/apache/spark/memory/StorageMemoryPool.scala
+++ b/core/src/main/scala/org/apache/spark/memory/StorageMemoryPool.scala
@@ -20,7 +20,8 @@ package org.apache.spark.memory
import javax.annotation.concurrent.GuardedBy
import org.apache.spark.SparkException
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.storage.BlockId
import org.apache.spark.storage.memory.MemoryStore
@@ -104,8 +105,8 @@ private[memory] class StorageMemoryPool(
def releaseMemory(size: Long): Unit = lock.synchronized {
if (size > _memoryUsed) {
- logWarning(s"Attempted to release $size bytes of storage " +
- s"memory when we only have ${_memoryUsed} bytes")
+ logWarning(log"Attempted to release ${MDC(NUM_BYTES, size)} bytes of storage " +
+ log"memory when we only have ${MDC(NUM_BYTES_USED, _memoryUsed)} bytes")
_memoryUsed = 0
} else {
_memoryUsed -= size
diff --git a/core/src/main/scala/org/apache/spark/metrics/ExecutorMetricType.scala b/core/src/main/scala/org/apache/spark/metrics/ExecutorMetricType.scala
index 50b7ddcb13ae..965468ac2418 100644
--- a/core/src/main/scala/org/apache/spark/metrics/ExecutorMetricType.scala
+++ b/core/src/main/scala/org/apache/spark/metrics/ExecutorMetricType.scala
@@ -24,7 +24,8 @@ import scala.jdk.CollectionConverters._
import org.apache.spark.SparkEnv
import org.apache.spark.executor.ProcfsMetricsGetter
-import org.apache.spark.internal.{config, Logging}
+import org.apache.spark.internal.{config, Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.memory.MemoryManager
/**
@@ -157,10 +158,13 @@ case object GarbageCollectionMetrics extends ExecutorMetricType with Logging {
} else if (!nonBuiltInCollectors.contains(mxBean.getName)) {
nonBuiltInCollectors = mxBean.getName +: nonBuiltInCollectors
// log it when first seen
- logWarning(s"To enable non-built-in garbage collector(s) " +
- s"$nonBuiltInCollectors, users should configure it(them) to " +
- s"${config.EVENT_LOG_GC_METRICS_YOUNG_GENERATION_GARBAGE_COLLECTORS.key} or " +
- s"${config.EVENT_LOG_GC_METRICS_OLD_GENERATION_GARBAGE_COLLECTORS.key}")
+ val youngGenerationGc = MDC(YOUNG_GENERATION_GC,
+ config.EVENT_LOG_GC_METRICS_YOUNG_GENERATION_GARBAGE_COLLECTORS.key)
+ val oldGenerationGc = MDC(OLD_GENERATION_GC,
+ config.EVENT_LOG_GC_METRICS_OLD_GENERATION_GARBAGE_COLLECTORS.key)
+ logWarning(log"To enable non-built-in garbage collector(s) " +
+ log"${MDC(NON_BUILT_IN_CONNECTORS, nonBuiltInCollectors)}, " +
+ log"users should configure it(them) to $youngGenerationGc or $oldGenerationGc")
} else {
// do nothing
}
diff --git a/core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala b/core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala
index 6f80a5cd6a06..709ce0060e15 100644
--- a/core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala
+++ b/core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala
@@ -27,7 +27,7 @@ import org.eclipse.jetty.servlet.ServletContextHandler
import org.apache.spark.{SecurityManager, SparkConf}
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.CLASS_NAME
+import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.config._
import org.apache.spark.metrics.sink.{MetricsServlet, PrometheusServlet, Sink}
import org.apache.spark.metrics.source.{Source, StaticSources}
@@ -141,12 +141,13 @@ private[spark] class MetricsSystem private (
// Only Driver and Executor set spark.app.id and spark.executor.id.
// Other instance types, e.g. Master and Worker, are not related to a specific application.
if (metricsNamespace.isEmpty) {
- logWarning(s"Using default name $defaultName for source because neither " +
- s"${METRICS_NAMESPACE.key} nor spark.app.id is set.")
+ logWarning(log"Using default name ${MDC(LogKeys.DEFAULT_NAME, defaultName)} " +
+ log"for source because neither " +
+ log"${MDC(LogKeys.CONFIG, METRICS_NAMESPACE.key)} nor spark.app.id is set.")
}
if (executorId.isEmpty) {
- logWarning(s"Using default name $defaultName for source because spark.executor.id is " +
- s"not set.")
+ logWarning(log"Using default name ${MDC(LogKeys.DEFAULT_NAME, defaultName)} " +
+ log"for source because spark.executor.id is not set.")
}
defaultName
}
@@ -189,7 +190,8 @@ private[spark] class MetricsSystem private (
registerSource(source)
} catch {
case e: Exception =>
- logError(log"Source class ${MDC(CLASS_NAME, classPath)} cannot be instantiated", e)
+ logError(log"Source class ${MDC(LogKeys.CLASS_NAME, classPath)} " +
+ log"cannot be instantiated", e)
}
}
}
@@ -229,7 +231,8 @@ private[spark] class MetricsSystem private (
}
} catch {
case e: Exception =>
- logError(log"Sink class ${MDC(CLASS_NAME, classPath)} cannot be instantiated")
+ logError(log"Sink class ${MDC(LogKeys.CLASS_NAME, classPath)} " +
+ log"cannot be instantiated")
throw e
}
}
diff --git a/core/src/main/scala/org/apache/spark/network/netty/NettyBlockRpcServer.scala b/core/src/main/scala/org/apache/spark/network/netty/NettyBlockRpcServer.scala
index aa0da153f7fa..a922eb336c28 100644
--- a/core/src/main/scala/org/apache/spark/network/netty/NettyBlockRpcServer.scala
+++ b/core/src/main/scala/org/apache/spark/network/netty/NettyBlockRpcServer.scala
@@ -23,7 +23,8 @@ import scala.jdk.CollectionConverters._
import scala.reflect.ClassTag
import org.apache.spark.SparkException
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.network.BlockDataManager
import org.apache.spark.network.buffer.NioManagedBuffer
import org.apache.spark.network.client.{RpcResponseCallback, StreamCallbackWithID, TransportClient}
@@ -55,17 +56,21 @@ class NettyBlockRpcServer(
BlockTransferMessage.Decoder.fromByteBuffer(rpcMessage)
} catch {
case e: IllegalArgumentException if e.getMessage.startsWith("Unknown message type") =>
- logWarning(s"This could be a corrupted RPC message (capacity: ${rpcMessage.capacity()}) " +
- s"from ${client.getSocketAddress}. Please use `spark.authenticate.*` configurations " +
- "in case of security incidents.")
+ logWarning(log"This could be a corrupted RPC message (capacity: " +
+ log"${MDC(RPC_MESSAGE_CAPACITY, rpcMessage.capacity())}) " +
+ log"from ${MDC(SOCKET_ADDRESS, client.getSocketAddress)}. " +
+ log"Please use `spark.authenticate.*` configurations " +
+ log"in case of security incidents.")
throw e
case _: IndexOutOfBoundsException | _: NegativeArraySizeException =>
// Netty may throw non-'IOException's for corrupted buffers. In this case,
// we ignore the entire message with warnings because we cannot trust any contents.
- logWarning(s"Ignored a corrupted RPC message (capacity: ${rpcMessage.capacity()}) " +
- s"from ${client.getSocketAddress}. Please use `spark.authenticate.*` configurations " +
- "in case of security incidents.")
+ logWarning(log"Ignored a corrupted RPC message (capacity: " +
+ log"${MDC(RPC_MESSAGE_CAPACITY, rpcMessage.capacity())}) " +
+ log"from ${MDC(SOCKET_ADDRESS, client.getSocketAddress)}. " +
+ log"Please use `spark.authenticate.*` configurations " +
+ log"in case of security incidents.")
return
}
logTrace(s"Received request: $message")
diff --git a/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala b/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
index 908ce1b233c5..ff899a2e56dc 100644
--- a/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
@@ -37,7 +37,8 @@ import org.apache.spark.annotation.DeveloperApi
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.errors.SparkCoreErrors
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.rdd.HadoopRDD.HadoopMapPartitionsWithSplitRDD
import org.apache.spark.scheduler.{HDFSCacheTaskLocation, HostTaskLocation}
@@ -239,11 +240,13 @@ class HadoopRDD[K, V](
if (fileSplit.getLength > conf.get(IO_WARNING_LARGEFILETHRESHOLD)) {
val codecFactory = new CompressionCodecFactory(jobConf)
if (Utils.isFileSplittable(path, codecFactory)) {
- logWarning(s"Loading one large file ${path.toString} with only one partition, " +
- s"we can increase partition numbers for improving performance.")
+ logWarning(log"Loading one large file ${MDC(PATH, path.toString)} " +
+ log"with only one partition, " +
+ log"we can increase partition numbers for improving performance.")
} else {
- logWarning(s"Loading one large unsplittable file ${path.toString} with only one " +
- s"partition, because the file is compressed by unsplittable compression codec.")
+ logWarning(log"Loading one large unsplittable file ${MDC(PATH, path.toString)} " +
+ log"with only one " +
+ log"partition, because the file is compressed by unsplittable compression codec.")
}
}
}
@@ -254,8 +257,8 @@ class HadoopRDD[K, V](
array
} catch {
case e: InvalidInputException if ignoreMissingFiles =>
- logWarning(s"${jobConf.get(FileInputFormat.INPUT_DIR)} doesn't exist and no" +
- s" partitions returned from this path.", e)
+ logWarning(log"${MDC(PATH, jobConf.get(FileInputFormat.INPUT_DIR))} " +
+ log"doesn't exist and no partitions returned from this path.", e)
Array.empty[Partition]
case e: IOException if e.getMessage.startsWith("Not a file:") =>
val path = e.getMessage.split(":").map(_.trim).apply(2)
@@ -310,13 +313,14 @@ class HadoopRDD[K, V](
inputFormat.getRecordReader(split.inputSplit.value, jobConf, Reporter.NULL)
} catch {
case e: FileNotFoundException if ignoreMissingFiles =>
- logWarning(s"Skipped missing file: ${split.inputSplit}", e)
+ logWarning(log"Skipped missing file: ${MDC(PATH, split.inputSplit)}", e)
finished = true
null
// Throw FileNotFoundException even if `ignoreCorruptFiles` is true
case e: FileNotFoundException if !ignoreMissingFiles => throw e
case e: IOException if ignoreCorruptFiles =>
- logWarning(s"Skipped the rest content in the corrupted file: ${split.inputSplit}", e)
+ logWarning(log"Skipped the rest content in the corrupted file: " +
+ log"${MDC(PATH, split.inputSplit)}", e)
finished = true
null
}
@@ -336,12 +340,13 @@ class HadoopRDD[K, V](
finished = !reader.next(key, value)
} catch {
case e: FileNotFoundException if ignoreMissingFiles =>
- logWarning(s"Skipped missing file: ${split.inputSplit}", e)
+ logWarning(log"Skipped missing file: ${MDC(PATH, split.inputSplit)}", e)
finished = true
// Throw FileNotFoundException even if `ignoreCorruptFiles` is true
case e: FileNotFoundException if !ignoreMissingFiles => throw e
case e: IOException if ignoreCorruptFiles =>
- logWarning(s"Skipped the rest content in the corrupted file: ${split.inputSplit}", e)
+ logWarning(log"Skipped the rest content in the corrupted file: " +
+ log"${MDC(PATH, split.inputSplit)}", e)
finished = true
}
if (!finished) {
diff --git a/core/src/main/scala/org/apache/spark/rdd/NewHadoopRDD.scala b/core/src/main/scala/org/apache/spark/rdd/NewHadoopRDD.scala
index 7db8531e4a59..a87d02287302 100644
--- a/core/src/main/scala/org/apache/spark/rdd/NewHadoopRDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/NewHadoopRDD.scala
@@ -37,7 +37,8 @@ import org.apache.spark._
import org.apache.spark.annotation.DeveloperApi
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.errors.SparkCoreErrors
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.rdd.NewHadoopRDD.NewHadoopMapPartitionsWithSplitRDD
import org.apache.spark.storage.StorageLevel
@@ -168,11 +169,13 @@ class NewHadoopRDD[K, V](
if (fileSplit.getLength > conf.get(IO_WARNING_LARGEFILETHRESHOLD)) {
val codecFactory = new CompressionCodecFactory(_conf)
if (Utils.isFileSplittable(path, codecFactory)) {
- logWarning(s"Loading one large file ${path.toString} with only one partition, " +
- s"we can increase partition numbers for improving performance.")
+ logWarning(log"Loading one large file ${MDC(PATH, path.toString)} " +
+ log"with only one partition, " +
+ log"we can increase partition numbers for improving performance.")
} else {
- logWarning(s"Loading one large unsplittable file ${path.toString} with only one " +
- s"partition, because the file is compressed by unsplittable compression codec.")
+ logWarning(log"Loading one large unsplittable file ${MDC(PATH, path.toString)} " +
+ log"with only one " +
+ log"partition, because the file is compressed by unsplittable compression codec.")
}
}
}
@@ -185,8 +188,8 @@ class NewHadoopRDD[K, V](
result
} catch {
case e: InvalidInputException if ignoreMissingFiles =>
- logWarning(s"${_conf.get(FileInputFormat.INPUT_DIR)} doesn't exist and no" +
- s" partitions returned from this path.", e)
+ logWarning(log"${MDC(PATH, _conf.get(FileInputFormat.INPUT_DIR))} " +
+ log"doesn't exist and no partitions returned from this path.", e)
Array.empty[Partition]
}
}
@@ -244,14 +247,15 @@ class NewHadoopRDD[K, V](
_reader
} catch {
case e: FileNotFoundException if ignoreMissingFiles =>
- logWarning(s"Skipped missing file: ${split.serializableHadoopSplit}", e)
+ logWarning(log"Skipped missing file: ${MDC(PATH, split.serializableHadoopSplit)}", e)
finished = true
null
// Throw FileNotFoundException even if `ignoreCorruptFiles` is true
case e: FileNotFoundException if !ignoreMissingFiles => throw e
case e: IOException if ignoreCorruptFiles =>
logWarning(
- s"Skipped the rest content in the corrupted file: ${split.serializableHadoopSplit}",
+ log"Skipped the rest content in the corrupted file: " +
+ log"${MDC(PATH, split.serializableHadoopSplit)}",
e)
finished = true
null
@@ -273,13 +277,14 @@ class NewHadoopRDD[K, V](
finished = !reader.nextKeyValue
} catch {
case e: FileNotFoundException if ignoreMissingFiles =>
- logWarning(s"Skipped missing file: ${split.serializableHadoopSplit}", e)
+ logWarning(log"Skipped missing file: ${MDC(PATH, split.serializableHadoopSplit)}", e)
finished = true
// Throw FileNotFoundException even if `ignoreCorruptFiles` is true
case e: FileNotFoundException if !ignoreMissingFiles => throw e
case e: IOException if ignoreCorruptFiles =>
logWarning(
- s"Skipped the rest content in the corrupted file: ${split.serializableHadoopSplit}",
+ log"Skipped the rest content in the corrupted file: " +
+ log"${MDC(PATH, split.serializableHadoopSplit)}",
e)
finished = true
}
diff --git a/core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala b/core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala
index f5a731d134ea..c0966dd5ede1 100644
--- a/core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala
@@ -35,7 +35,8 @@ import org.apache.hadoop.mapreduce.{Job => NewAPIHadoopJob, OutputFormat => NewO
import org.apache.spark._
import org.apache.spark.Partitioner.defaultPartitioner
import org.apache.spark.errors.SparkCoreErrors
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.SPECULATION_ENABLED
import org.apache.spark.internal.io._
import org.apache.spark.partial.{BoundedDouble, PartialResult}
@@ -1051,10 +1052,11 @@ class PairRDDFunctions[K, V](self: RDD[(K, V)])
val outputCommitterClass = hadoopConf.get("mapred.output.committer.class", "")
if (speculationEnabled && outputCommitterClass.contains("Direct")) {
val warningMessage =
- s"$outputCommitterClass may be an output committer that writes data directly to " +
- "the final location. Because speculation is enabled, this output committer may " +
- "cause data loss (see the case in SPARK-10063). If possible, please use an output " +
- "committer that does not have this behavior (e.g. FileOutputCommitter)."
+ log"${MDC(CLASS_NAME, outputCommitterClass)} " +
+ log"may be an output committer that writes data directly to " +
+ log"the final location. Because speculation is enabled, this output committer may " +
+ log"cause data loss (see the case in SPARK-10063). If possible, please use an output " +
+ log"committer that does not have this behavior (e.g. FileOutputCommitter)."
logWarning(warningMessage)
}
diff --git a/core/src/main/scala/org/apache/spark/rdd/RDD.scala b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
index a48eaa253ad1..1f44f7e782c4 100644
--- a/core/src/main/scala/org/apache/spark/rdd/RDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
@@ -36,7 +36,8 @@ import org.apache.spark.Partitioner._
import org.apache.spark.annotation.{DeveloperApi, Experimental, Since}
import org.apache.spark.api.java.JavaRDD
import org.apache.spark.errors.SparkCoreErrors
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.RDD_LIMIT_SCALE_UP_FACTOR
import org.apache.spark.partial.BoundedDouble
@@ -643,7 +644,8 @@ abstract class RDD[T: ClassTag](
// this shouldn't happen often because we use a big multiplier for the initial size
var numIters = 0
while (samples.length < num) {
- logWarning(s"Needed to re-sample due to insufficient sample size. Repeat #$numIters")
+ logWarning(log"Needed to re-sample due to insufficient sample size. " +
+ log"Repeat #${MDC(NUM_ITERATIONS, numIters)}")
samples = this.sample(withReplacement, fraction, rand.nextInt()).collect()
numIters += 1
}
diff --git a/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala b/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala
index 863bcd5b12d3..4bd2de62d75c 100644
--- a/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala
@@ -29,7 +29,8 @@ import org.apache.hadoop.fs.Path
import org.apache.spark._
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.errors.SparkCoreErrors
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.{BUFFER_SIZE, CACHE_CHECKPOINT_PREFERRED_LOCS_EXPIRE_TIME, CHECKPOINT_COMPRESS}
import org.apache.spark.io.CompressionCodec
import org.apache.spark.util.{SerializableConfiguration, Utils}
@@ -234,7 +235,7 @@ private[spark] object ReliableCheckpointRDD extends Logging {
// Some other copy of this task must've finished before us and renamed it
logInfo(s"Final output path $finalOutputPath already exists; not overwriting it")
if (!fs.delete(tempOutputPath, false)) {
- logWarning(s"Error deleting ${tempOutputPath}")
+ logWarning(log"Error deleting ${MDC(PATH, tempOutputPath)}")
}
}
}
@@ -261,7 +262,8 @@ private[spark] object ReliableCheckpointRDD extends Logging {
logDebug(s"Written partitioner to $partitionerFilePath")
} catch {
case NonFatal(e) =>
- logWarning(s"Error writing partitioner $partitioner to $checkpointDirPath")
+ logWarning(log"Error writing partitioner ${MDC(PARTITIONER, partitioner)} to " +
+ log"${MDC(PATH, checkpointDirPath)}")
}
}
@@ -298,8 +300,8 @@ private[spark] object ReliableCheckpointRDD extends Logging {
logDebug("No partitioner file", e)
None
case NonFatal(e) =>
- logWarning(s"Error reading partitioner from $checkpointDirPath, " +
- s"partitioner will not be recovered which may lead to performance loss", e)
+ logWarning(log"Error reading partitioner from ${MDC(PATH, checkpointDirPath)}, " +
+ log"partitioner will not be recovered which may lead to performance loss", e)
None
}
}
diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala
index e95dbe973691..adc1eeeb3127 100644
--- a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala
+++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala
@@ -26,7 +26,8 @@ import scala.jdk.CollectionConverters._
import org.apache.spark.{SparkConf, SparkContext, SparkEnv, SparkException}
import org.apache.spark.annotation.{Evolving, Since}
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.Python.PYSPARK_EXECUTOR_MEMORY
import org.apache.spark.util.Utils
@@ -221,8 +222,8 @@ class ResourceProfile(
}
taskResourcesToCheck -= rName
} else {
- logWarning(s"The executor resource config for resource: $rName was specified but " +
- "no corresponding task resource request was specified.")
+ logWarning(log"The executor resource config for resource: ${MDC(RESOURCE_NAME, rName)} " +
+ log"was specified but no corresponding task resource request was specified.")
}
}
if (taskResourcesToCheck.nonEmpty) {
diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceUtils.scala b/core/src/main/scala/org/apache/spark/resource/ResourceUtils.scala
index 8718ce8ea083..78c45cdc7541 100644
--- a/core/src/main/scala/org/apache/spark/resource/ResourceUtils.scala
+++ b/core/src/main/scala/org/apache/spark/resource/ResourceUtils.scala
@@ -28,7 +28,8 @@ import org.json4s.jackson.JsonMethods._
import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.annotation.DeveloperApi
import org.apache.spark.api.resource.ResourceDiscoveryPlugin
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.{EXECUTOR_CORES, RESOURCES_DISCOVERY_PLUGIN, SPARK_TASK_PREFIX}
import org.apache.spark.internal.config.Tests.RESOURCES_WARNING_TESTING
import org.apache.spark.util.ArrayImplicits._
@@ -454,13 +455,15 @@ private[spark] object ResourceUtils extends Logging {
if (limitingResource.nonEmpty && !limitingResource.equals(ResourceProfile.CPUS)) {
if ((taskCpus * maxTaskPerExec) < cores) {
val resourceNumSlots = Math.floor(cores/taskCpus).toInt
- val message = s"The configuration of cores (exec = ${cores} " +
- s"task = ${taskCpus}, runnable tasks = ${resourceNumSlots}) will " +
- s"result in wasted resources due to resource ${limitingResource} limiting the " +
- s"number of runnable tasks per executor to: ${maxTaskPerExec}. Please adjust " +
- "your configuration."
+ val message = log"The configuration of cores (exec = ${MDC(NUM_CORES, cores)} " +
+ log"task = ${MDC(NUM_TASK_CPUS, taskCpus)}, runnable tasks = " +
+ log"${MDC(NUM_RESOURCE_SLOTS, resourceNumSlots)}) will " +
+ log"result in wasted resources due to resource ${MDC(RESOURCE, limitingResource)} " +
+ log"limiting the number of runnable tasks per executor to: " +
+ log"${MDC(NUM_TASKS, maxTaskPerExec)}. Please adjust " +
+ log"your configuration."
if (sparkConf.get(RESOURCES_WARNING_TESTING)) {
- throw new SparkException(message)
+ throw new SparkException(message.message)
} else {
logWarning(message)
}
@@ -476,14 +479,16 @@ private[spark] object ResourceUtils extends Logging {
val origTaskAmount = treq.amount
val taskReqStr = s"${origTaskAmount}/${numParts}"
val resourceNumSlots = (execAmount * numParts / taskAmount).toInt
- val message = s"The configuration of resource: ${treq.resourceName} " +
- s"(exec = ${execAmount}, task = ${taskReqStr}, " +
- s"runnable tasks = ${resourceNumSlots}) will " +
- s"result in wasted resources due to resource ${limitingResource} limiting the " +
- s"number of runnable tasks per executor to: ${maxTaskPerExec}. Please adjust " +
- "your configuration."
+ val message = log"The configuration of resource: " +
+ log"${MDC(RESOURCE_NAME, treq.resourceName)} " +
+ log"(exec = ${MDC(EXEC_AMOUNT, execAmount)}, " +
+ log"task = ${MDC(TASK_REQUIREMENTS, taskReqStr)}, " +
+ log"runnable tasks = ${MDC(NUM_RESOURCE_SLOTS, resourceNumSlots)}) will " +
+ log"result in wasted resources due to resource ${MDC(RESOURCE, limitingResource)} " +
+ log"limiting the number of runnable tasks per executor to: " +
+ log"${MDC(NUM_TASKS, maxTaskPerExec)}. Please adjust your configuration."
if (sparkConf.get(RESOURCES_WARNING_TESTING)) {
- throw new SparkException(message)
+ throw new SparkException(message.message)
} else {
logWarning(message)
}
diff --git a/core/src/main/scala/org/apache/spark/rpc/netty/Dispatcher.scala b/core/src/main/scala/org/apache/spark/rpc/netty/Dispatcher.scala
index 0e35842fece9..8acfef38659c 100644
--- a/core/src/main/scala/org/apache/spark/rpc/netty/Dispatcher.scala
+++ b/core/src/main/scala/org/apache/spark/rpc/netty/Dispatcher.scala
@@ -25,7 +25,8 @@ import scala.jdk.CollectionConverters._
import scala.util.control.NonFatal
import org.apache.spark.{SparkEnv, SparkException}
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.network.client.RpcResponseCallback
import org.apache.spark.rpc._
@@ -123,7 +124,8 @@ private[netty] class Dispatcher(nettyEnv: NettyRpcEnv, numUsableCores: Int) exte
val name = iter.next
postMessage(name, message, (e) => { e match {
case e: RpcEnvStoppedException => logDebug(s"Message $message dropped. ${e.getMessage}")
- case e: Throwable => logWarning(s"Message $message dropped. ${e.getMessage}")
+ case e: Throwable =>
+ logWarning(log"Message ${MDC(MESSAGE, message)} dropped. ${MDC(ERROR, e.getMessage)}")
}}
)}
}
@@ -154,7 +156,8 @@ private[netty] class Dispatcher(nettyEnv: NettyRpcEnv, numUsableCores: Int) exte
// cluster in spark shell.
case re: RpcEnvStoppedException => logDebug(s"Message $message dropped. ${re.getMessage}")
case e if SparkEnv.get.isStopped =>
- logWarning(s"Message $message dropped due to sparkEnv is stopped. ${e.getMessage}")
+ logWarning(log"Message ${MDC(MESSAGE, message)} dropped due to sparkEnv " +
+ log"is stopped. ${MDC(ERROR, e.getMessage)}")
case e => throw e
})
}
diff --git a/core/src/main/scala/org/apache/spark/rpc/netty/Inbox.scala b/core/src/main/scala/org/apache/spark/rpc/netty/Inbox.scala
index be503d8e564f..0de67a65593b 100644
--- a/core/src/main/scala/org/apache/spark/rpc/netty/Inbox.scala
+++ b/core/src/main/scala/org/apache/spark/rpc/netty/Inbox.scala
@@ -23,7 +23,7 @@ import scala.util.control.NonFatal
import org.apache.spark.SparkException
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.END_POINT
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.rpc.{RpcAddress, RpcEndpoint, ThreadSafeRpcEndpoint}
@@ -194,7 +194,8 @@ private[netty] class Inbox(val endpointName: String, val endpoint: RpcEndpoint)
* Exposed for testing.
*/
protected def onDrop(message: InboxMessage): Unit = {
- logWarning(s"Drop $message because endpoint $endpointName is stopped")
+ logWarning(log"Drop ${MDC(MESSAGE, message)} " +
+ log"because endpoint ${MDC(END_POINT, endpointName)} is stopped")
}
/**
diff --git a/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala b/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala
index 7909f2327cdf..c2688610fe8b 100644
--- a/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala
+++ b/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala
@@ -30,7 +30,8 @@ import scala.util.{DynamicVariable, Failure, Success, Try}
import scala.util.control.NonFatal
import org.apache.spark.{SecurityManager, SparkConf, SparkContext}
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.EXECUTOR_ID
import org.apache.spark.internal.config.Network._
import org.apache.spark.network.TransportContext
@@ -216,7 +217,7 @@ private[netty] class NettyRpcEnv(
if (!promise.tryFailure(e)) {
e match {
case e : RpcEnvStoppedException => logDebug(s"Ignored failure: $e")
- case _ => logWarning(s"Ignored failure: $e")
+ case _ => logWarning(log"Ignored failure: ${MDC(ERROR, e)}")
}
}
}
@@ -225,7 +226,7 @@ private[netty] class NettyRpcEnv(
case RpcFailure(e) => onFailure(e)
case rpcReply =>
if (!promise.trySuccess(rpcReply)) {
- logWarning(s"Ignored message: $reply")
+ logWarning(log"Ignored message: ${MDC(MESSAGE, reply)}")
}
}
diff --git a/core/src/main/scala/org/apache/spark/rpc/netty/Outbox.scala b/core/src/main/scala/org/apache/spark/rpc/netty/Outbox.scala
index 205e6e966866..b212a818ffc4 100644
--- a/core/src/main/scala/org/apache/spark/rpc/netty/Outbox.scala
+++ b/core/src/main/scala/org/apache/spark/rpc/netty/Outbox.scala
@@ -46,7 +46,7 @@ private[netty] case class OneWayOutboxMessage(content: ByteBuffer) extends Outbo
override def onFailure(e: Throwable): Unit = {
e match {
case e1: RpcEnvStoppedException => logDebug(e1.getMessage)
- case e1: Throwable => logWarning(s"Failed to send one-way RPC.", e1)
+ case e1: Throwable => logWarning(log"Failed to send one-way RPC.", e1)
}
}
diff --git a/core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala b/core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala
index 5184a76847d8..16e9211b5485 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala
@@ -24,7 +24,7 @@ import com.codahale.metrics.{Gauge, Timer}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.EVENT_QUEUE
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.util.Utils
@@ -187,8 +187,9 @@ private class AsyncEventQueue(
if (lastReportTimestamp.compareAndSet(lastReportTime, curTime)) {
val previous = new java.util.Date(lastReportTime)
lastDroppedEventsCounter = droppedEventsCount
- logWarning(s"Dropped $droppedCountIncreased events from $name since " +
- s"${if (lastReportTime == 0) "the application started" else s"$previous"}.")
+ logWarning(log"Dropped ${MDC(NUM_EVENTS, droppedCountIncreased)} events from " +
+ log"${MDC(EVENT_NAME, name)} since " +
+ (if (lastReportTime == 0) log"the application started" else log"${MDC(TIME, previous)}"))
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index bc64c297324c..cc9ae5eb1ebe 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -37,7 +37,7 @@ import org.apache.spark.broadcast.Broadcast
import org.apache.spark.errors.SparkCoreErrors
import org.apache.spark.executor.{ExecutorMetrics, TaskMetrics}
import org.apache.spark.internal.{config, Logging, MDC}
-import org.apache.spark.internal.LogKeys.{ACCUMULATOR_ID, CLASS_NAME, JOB_ID, PARTITION_ID, STAGE_ID, TASK_ID}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.{LEGACY_ABORT_STAGE_AFTER_KILL_TASKS, RDD_CACHE_VISIBILITY_TRACKING_ENABLED}
import org.apache.spark.internal.config.Tests.TEST_NO_STAGE_RETRY
import org.apache.spark.network.shuffle.{BlockStoreClient, MergeFinalizerListener}
@@ -1221,7 +1221,8 @@ private[spark] class DAGScheduler(
}
}
if (activeInGroup.isEmpty && !cancelFutureJobs) {
- logWarning(s"Failed to cancel job group $groupId. Cannot find active jobs for it.")
+ logWarning(log"Failed to cancel job group ${MDC(GROUP_ID, groupId)}. " +
+ log"Cannot find active jobs for it.")
}
val jobIds = activeInGroup.map(_.jobId)
jobIds.foreach(handleJobCancellation(_,
@@ -1328,9 +1329,11 @@ private[spark] class DAGScheduler(
val numCheckFailures = barrierJobIdToNumTasksCheckFailures.compute(jobId,
(_: Int, value: Int) => value + 1)
- logWarning(s"Barrier stage in job $jobId requires ${e.requiredConcurrentTasks} slots, " +
- s"but only ${e.maxConcurrentTasks} are available. " +
- s"Will retry up to ${maxFailureNumTasksCheck - numCheckFailures + 1} more times")
+ logWarning(log"Barrier stage in job ${MDC(JOB_ID, jobId)} " +
+ log"requires ${MDC(NUM_SLOTS, e.requiredConcurrentTasks)} slots, " +
+ log"but only ${MDC(MAX_SLOTS, e.maxConcurrentTasks)} are available. " +
+ log"Will retry up to ${MDC(NUM_RETRIES, maxFailureNumTasksCheck - numCheckFailures + 1)} " +
+ log"more times")
if (numCheckFailures <= maxFailureNumTasksCheck) {
messageScheduler.schedule(
@@ -1350,7 +1353,7 @@ private[spark] class DAGScheduler(
}
case e: Exception =>
- logWarning("Creating new stage failed due to exception - job: " + jobId, e)
+ logWarning(log"Creating new stage failed due to exception - job: ${MDC(JOB_ID, jobId)}", e)
listener.jobFailed(e)
return
}
@@ -1393,7 +1396,7 @@ private[spark] class DAGScheduler(
finalStage = getOrCreateShuffleMapStage(dependency, jobId)
} catch {
case e: Exception =>
- logWarning("Creating new stage failed due to exception - job: " + jobId, e)
+ logWarning(log"Creating new stage failed due to exception - job: ${MDC(JOB_ID, jobId)}", e)
listener.jobFailed(e)
return
}
@@ -1622,8 +1625,8 @@ private[spark] class DAGScheduler(
}
if (taskBinaryBytes.length > TaskSetManager.TASK_SIZE_TO_WARN_KIB * 1024) {
- logWarning(s"Broadcasting large task binary with size " +
- s"${Utils.bytesToString(taskBinaryBytes.length)}")
+ logWarning(log"Broadcasting large task binary with size " +
+ log"${MDC(NUM_BYTES, Utils.bytesToString(taskBinaryBytes.length))}")
}
taskBinary = sc.broadcast(taskBinaryBytes)
} catch {
@@ -1791,8 +1794,10 @@ private[spark] class DAGScheduler(
shouldInterruptThread.toBoolean
} catch {
case e: IllegalArgumentException =>
- logWarning(s"${SparkContext.SPARK_JOB_INTERRUPT_ON_CANCEL} in Job ${job.jobId} " +
- s"is invalid: $shouldInterruptThread. Using 'false' instead", e)
+ logWarning(log"${MDC(CONFIG, SparkContext.SPARK_JOB_INTERRUPT_ON_CANCEL)} " +
+ log"in Job ${MDC(JOB_ID, job.jobId)} " +
+ log"is invalid: ${MDC(CONFIG2, shouldInterruptThread)}. " +
+ log"Using 'false' instead", e)
false
}
}
@@ -1931,7 +1936,8 @@ private[spark] class DAGScheduler(
reason = "Stage finished")
} catch {
case e: UnsupportedOperationException =>
- logWarning(s"Could not cancel tasks for stage $stageId", e)
+ logWarning(log"Could not cancel tasks " +
+ log"for stage ${MDC(STAGE_ID, stageId)}", e)
}
listenerBus.post(
SparkListenerJobEnd(job.jobId, clock.getTimeMillis(), JobSucceeded))
@@ -2236,7 +2242,7 @@ private[spark] class DAGScheduler(
case e: UnsupportedOperationException =>
// Cannot continue with barrier stage if failed to cancel zombie barrier tasks.
// TODO SPARK-24877 leave the zombie tasks and ignore their completion events.
- logWarning(s"Could not kill all tasks for stage $stageId", e)
+ logWarning(log"Could not kill all tasks for stage ${MDC(STAGE_ID, stageId)}", e)
abortStage(failedStage, "Could not kill zombie barrier tasks for stage " +
s"$failedStage (${failedStage.name})", Some(e))
}
@@ -2451,8 +2457,9 @@ private[spark] class DAGScheduler(
}
override def onShuffleMergeFailure(e: Throwable): Unit = {
- logWarning(s"Exception encountered when trying to finalize shuffle " +
- s"merge on ${shuffleServiceLoc.host} for shuffle $shuffleId", e)
+ logWarning(log"Exception encountered when trying to finalize shuffle " +
+ log"merge on ${MDC(HOST_PORT, shuffleServiceLoc.host)} " +
+ log"for shuffle ${MDC(SHUFFLE_ID, shuffleId)}", e)
// Do not fail the future as this would cause dag scheduler to prematurely
// give up on waiting for merge results from the remaining shuffle services
// if one fails
@@ -2892,7 +2899,7 @@ private[spark] class DAGScheduler(
markStageAsFinished(stage, Some(reason))
} catch {
case e: UnsupportedOperationException =>
- logWarning(s"Could not cancel tasks for stage $stageId", e)
+ logWarning(log"Could not cancel tasks for stage ${MDC(STAGE_ID, stageId)}", e)
ableToCancelStages = false
}
}
diff --git a/core/src/main/scala/org/apache/spark/scheduler/HealthTracker.scala b/core/src/main/scala/org/apache/spark/scheduler/HealthTracker.scala
index 45ae1df156e1..cecf5d498ac4 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/HealthTracker.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/HealthTracker.scala
@@ -23,7 +23,7 @@ import scala.collection.mutable.{ArrayBuffer, HashMap, HashSet}
import org.apache.spark.{ExecutorAllocationClient, SparkConf, SparkContext}
import org.apache.spark.internal.{config, Logging, MDC}
-import org.apache.spark.internal.LogKeys.HOST
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.util.{Clock, SystemClock, Utils}
/**
@@ -209,8 +209,9 @@ private[scheduler] class HealthTracker (
}
}
case None =>
- logWarning(s"Not attempting to kill executors on excluded host $node " +
- s"since allocation client is not defined.")
+ logWarning(
+ log"Not attempting to kill executors on excluded host ${MDC(HOST_PORT, node)} " +
+ log"since allocation client is not defined.")
}
}
}
@@ -437,10 +438,12 @@ private[spark] object HealthTracker extends Logging {
val legacyKey = config.EXCLUDE_ON_FAILURE_LEGACY_TIMEOUT_CONF.key
conf.get(config.EXCLUDE_ON_FAILURE_LEGACY_TIMEOUT_CONF).exists { legacyTimeout =>
if (legacyTimeout == 0) {
- logWarning(s"Turning off excludeOnFailure due to legacy configuration: $legacyKey == 0")
+ logWarning(log"Turning off excludeOnFailure due to legacy configuration: " +
+ log"${MDC(CONFIG, legacyKey)} == 0")
false
} else {
- logWarning(s"Turning on excludeOnFailure due to legacy configuration: $legacyKey > 0")
+ logWarning(log"Turning on excludeOnFailure due to legacy configuration: " +
+ log"${MDC(CONFIG, legacyKey)} > 0")
true
}
}
diff --git a/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala b/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala
index f0267ad35ad2..26c7486010c0 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/ReplayListenerBus.scala
@@ -25,7 +25,7 @@ import com.fasterxml.jackson.core.JsonParseException
import com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.{LINE, LINE_NUM, PATH}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.scheduler.ReplayListenerBus._
import org.apache.spark.util.JsonProtocol
@@ -92,7 +92,7 @@ private[spark] class ReplayListenerBus extends SparkListenerBus with Logging {
// Ignore unknown events, parse through the event log file.
// To avoid spamming, warnings are only displayed once for each unknown event.
if (!unrecognizedEvents.contains(e.getMessage)) {
- logWarning(s"Drop unrecognized event: ${e.getMessage}")
+ logWarning(log"Drop unrecognized event: ${MDC(ERROR, e.getMessage)}")
unrecognizedEvents.add(e.getMessage)
}
logDebug(s"Drop incompatible event log: $currentLine")
@@ -100,7 +100,7 @@ private[spark] class ReplayListenerBus extends SparkListenerBus with Logging {
// Ignore unrecognized properties, parse through the event log file.
// To avoid spamming, warnings are only displayed once for each unrecognized property.
if (!unrecognizedProperties.contains(e.getMessage)) {
- logWarning(s"Drop unrecognized property: ${e.getMessage}")
+ logWarning(log"Drop unrecognized property: ${MDC(ERROR, e.getMessage)}")
unrecognizedProperties.add(e.getMessage)
}
logDebug(s"Drop incompatible event log: $currentLine")
@@ -111,8 +111,9 @@ private[spark] class ReplayListenerBus extends SparkListenerBus with Logging {
if (!maybeTruncated || lineEntries.hasNext) {
throw jpe
} else {
- logWarning(s"Got JsonParseException from log file $sourceName" +
- s" at line $lineNumber, the file might not have finished writing cleanly.")
+ logWarning(log"Got JsonParseException from log file ${MDC(FILE_NAME, sourceName)}" +
+ log" at line ${MDC(LINE_NUM, lineNumber)}, " +
+ log"the file might not have finished writing cleanly.")
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/scheduler/SchedulableBuilder.scala b/core/src/main/scala/org/apache/spark/scheduler/SchedulableBuilder.scala
index c14018960267..b5cc6261cea3 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/SchedulableBuilder.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/SchedulableBuilder.scala
@@ -27,7 +27,8 @@ import org.apache.hadoop.fs.Path
import org.apache.spark.SparkContext
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.PATH
+import org.apache.spark.internal.LogKeys
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.{SCHEDULER_ALLOCATION_FILE, SCHEDULER_MODE}
import org.apache.spark.scheduler.SchedulingMode.SchedulingMode
import org.apache.spark.util.Utils
@@ -154,9 +155,12 @@ private[spark] class FairSchedulableBuilder(val rootPool: Pool, sc: SparkContext
val xmlSchedulingMode =
(poolNode \ SCHEDULING_MODE_PROPERTY).text.trim.toUpperCase(Locale.ROOT)
- val warningMessage = s"Unsupported schedulingMode: $xmlSchedulingMode found in " +
- s"Fair Scheduler configuration file: $fileName, using " +
- s"the default schedulingMode: $defaultValue for pool: $poolName"
+ val warningMessage = log"Unsupported schedulingMode: " +
+ log"${MDC(XML_SCHEDULING_MODE, xmlSchedulingMode)} found in " +
+ log"Fair Scheduler configuration file: ${MDC(FILE_NAME, fileName)}, using " +
+ log"the default schedulingMode: " +
+ log"${MDC(LogKeys.DEFAULT_SCHEDULING_MODE, defaultValue)} for pool: " +
+ log"${MDC(POOL_NAME, poolName)}"
try {
if (SchedulingMode.withName(xmlSchedulingMode) != SchedulingMode.NONE) {
SchedulingMode.withName(xmlSchedulingMode)
@@ -165,7 +169,7 @@ private[spark] class FairSchedulableBuilder(val rootPool: Pool, sc: SparkContext
defaultValue
}
} catch {
- case e: NoSuchElementException =>
+ case _: NoSuchElementException =>
logWarning(warningMessage)
defaultValue
}
@@ -182,10 +186,12 @@ private[spark] class FairSchedulableBuilder(val rootPool: Pool, sc: SparkContext
try {
data.toInt
} catch {
- case e: NumberFormatException =>
- logWarning(s"Error while loading fair scheduler configuration from $fileName: " +
- s"$propertyName is blank or invalid: $data, using the default $propertyName: " +
- s"$defaultValue for pool: $poolName")
+ case _: NumberFormatException =>
+ logWarning(log"Error while loading fair scheduler configuration from " +
+ log"${MDC(FILE_NAME, fileName)}: " +
+ log"${MDC(PROPERTY_NAME, propertyName)} is blank or invalid: ${MDC(DATA, data)}, " +
+ log"using the default ${MDC(DEFAULT_NAME, propertyName)}: " +
+ log"${MDC(DEFAULT_VALUE, defaultValue)} for pool: ${MDC(POOL_NAME, poolName)}")
defaultValue
}
}
@@ -203,11 +209,15 @@ private[spark] class FairSchedulableBuilder(val rootPool: Pool, sc: SparkContext
parentPool = new Pool(poolName, DEFAULT_SCHEDULING_MODE,
DEFAULT_MINIMUM_SHARE, DEFAULT_WEIGHT)
rootPool.addSchedulable(parentPool)
- logWarning(s"A job was submitted with scheduler pool $poolName, which has not been " +
- "configured. This can happen when the file that pools are read from isn't set, or " +
- s"when that file doesn't contain $poolName. Created $poolName with default " +
- s"configuration (schedulingMode: $DEFAULT_SCHEDULING_MODE, " +
- s"minShare: $DEFAULT_MINIMUM_SHARE, weight: $DEFAULT_WEIGHT)")
+ logWarning(log"A job was submitted with scheduler pool " +
+ log"${MDC(SCHEDULER_POOL_NAME, poolName)}, which has not been " +
+ log"configured. This can happen when the file that pools are read from isn't set, or " +
+ log"when that file doesn't contain ${MDC(POOL_NAME, poolName)}. " +
+ log"Created ${MDC(CREATED_POOL_NAME, poolName)} with default " +
+ log"configuration (schedulingMode: " +
+ log"${MDC(LogKeys.DEFAULT_SCHEDULING_MODE, DEFAULT_SCHEDULING_MODE)}, " +
+ log"minShare: ${MDC(MIN_SHARE, DEFAULT_MINIMUM_SHARE)}, " +
+ log"weight: ${MDC(WEIGHT, DEFAULT_WEIGHT)}")
}
parentPool.addSchedulable(manager)
logInfo("Added task set " + manager.name + " tasks to pool " + poolName)
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
index bc2afd32b8a9..15bdd58288f1 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
@@ -34,6 +34,7 @@ import org.apache.spark.TaskState.TaskState
import org.apache.spark.errors.SparkCoreErrors
import org.apache.spark.executor.ExecutorMetrics
import org.apache.spark.internal.{config, Logging, LogKeys, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.resource.ResourceProfile
import org.apache.spark.rpc.RpcEndpoint
@@ -336,7 +337,8 @@ private[spark] class TaskSchedulerImpl(
backend.killTask(taskId, execId.get, interruptThread, reason)
true
} else {
- logWarning(s"Could not kill task $taskId because no task with that ID was found.")
+ logWarning(log"Could not kill task ${MDC(TASK_ID, taskId)} " +
+ log"because no task with that ID was found.")
false
}
}
@@ -662,14 +664,17 @@ private[spark] class TaskSchedulerImpl(
// always reject the offered resources. As a result, the barrier taskset can't get
// launched. And if we retry the resourceOffer, we'd go through the same path again
// and get into the endless loop in the end.
- val errorMsg = s"Fail resource offers for barrier stage ${taskSet.stageId} " +
- s"because only ${barrierPendingLaunchTasks.length} out of a total number " +
- s"of ${taskSet.numTasks} tasks got resource offers. We highly recommend " +
- "you to use the non-legacy delay scheduling by setting " +
- s"${LEGACY_LOCALITY_WAIT_RESET.key} to false to get rid of this error."
- logWarning(errorMsg)
- taskSet.abort(errorMsg)
- throw SparkCoreErrors.sparkError(errorMsg)
+ val logMsg = log"Fail resource offers for barrier stage " +
+ log"${MDC(STAGE_ID, taskSet.stageId)} because only " +
+ log"${MDC(NUM_PENDING_LAUNCH_TASKS, barrierPendingLaunchTasks.length)} " +
+ log"out of a total number " +
+ log"of ${MDC(NUM_TASKS, taskSet.numTasks)} tasks got resource offers. " +
+ log"We highly recommend you to use the non-legacy delay scheduling by setting " +
+ log"${MDC(CONFIG, LEGACY_LOCALITY_WAIT_RESET.key)} to false " +
+ log"to get rid of this error."
+ logWarning(logMsg)
+ taskSet.abort(logMsg.message)
+ throw SparkCoreErrors.sparkError(logMsg.message)
} else {
val curTime = clock.getTimeMillis()
if (curTime - taskSet.lastResourceOfferFailLogTime >
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
index 1d86ebc0a734..b7ff443231f3 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
@@ -31,6 +31,7 @@ import org.apache.spark.InternalAccumulator.{input, shuffleRead}
import org.apache.spark.TaskState.TaskState
import org.apache.spark.errors.SparkCoreErrors
import org.apache.spark.internal.{config, Logging, MDC}
+import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.scheduler.SchedulingMode._
@@ -542,9 +543,10 @@ private[spark] class TaskSetManager(
if (serializedTask.limit() > TaskSetManager.TASK_SIZE_TO_WARN_KIB * 1024 &&
!emittedTaskSizeWarning) {
emittedTaskSizeWarning = true
- logWarning(s"Stage ${task.stageId} contains a task of very large size " +
- s"(${serializedTask.limit() / 1024} KiB). The maximum recommended task size is " +
- s"${TaskSetManager.TASK_SIZE_TO_WARN_KIB} KiB.")
+ logWarning(log"Stage ${MDC(STAGE_ID, task.stageId)} contains a task of very large size " +
+ log"(${MDC(NUM_BYTES, serializedTask.limit() / 1024)} KiB). " +
+ log"The maximum recommended task size is " +
+ log"${MDC(NUM_BYTES_TO_WARN, TaskSetManager.TASK_SIZE_TO_WARN_KIB)} KiB.")
}
addRunningTask(taskId)
@@ -936,7 +938,11 @@ private[spark] class TaskSetManager(
copiesRunning(index) -= 1
var accumUpdates: Seq[AccumulatorV2[_, _]] = Seq.empty
var metricPeaks: Array[Long] = Array.empty
- val failureReason = s"Lost ${taskName(tid)} (${info.host} " +
+ val failureReason = log"Lost ${MDC(TASK_NAME, taskName(tid))} " +
+ log"(${MDC(HOST_PORT, info.host)} " +
+ log"executor ${MDC(LogKeys.EXECUTOR_ID, info.executorId)}): " +
+ log"${MDC(ERROR, reason.toErrorString)}"
+ val failureReasonString = s"Lost ${taskName(tid)} (${info.host} " +
s"executor ${info.executorId}): ${reason.toErrorString}"
val failureException: Option[Throwable] = reason match {
case fetchFailed: FetchFailed =>
@@ -1019,7 +1025,7 @@ private[spark] class TaskSetManager(
"maximum number of failures for the task.")
None
- case e: TaskFailedReason => // TaskResultLost and others
+ case _: TaskFailedReason => // TaskResultLost and others
logWarning(failureReason)
None
}
@@ -1034,13 +1040,13 @@ private[spark] class TaskSetManager(
if (!isZombie && reason.countTowardsTaskFailures) {
assert (null != failureReason)
taskSetExcludelistHelperOpt.foreach(_.updateExcludedForFailedTask(
- info.host, info.executorId, index, failureReason))
+ info.host, info.executorId, index, failureReasonString))
numFailures(index) += 1
if (numFailures(index) >= maxTaskFailures) {
logError(log"Task ${MDC(TASK_ID, index)} in stage ${MDC(STAGE_ID, taskSet.id)} failed " +
log"${MDC(MAX_ATTEMPTS, maxTaskFailures)} times; aborting job")
abort("Task %d in stage %s failed %d times, most recent failure: %s\nDriver stacktrace:"
- .format(index, taskSet.id, maxTaskFailures, failureReason), failureException)
+ .format(index, taskSet.id, maxTaskFailures, failureReasonString), failureException)
return
}
}
@@ -1281,7 +1287,8 @@ private[spark] class TaskSetManager(
if (foundTasks) {
val elapsedMs = clock.getTimeMillis() - timeMs
if (elapsedMs > minTimeToSpeculation) {
- logWarning(s"Time to checkSpeculatableTasks ${elapsedMs}ms > ${minTimeToSpeculation}ms")
+ logWarning(log"Time to checkSpeculatableTasks ${MDC(TIME_UNITS, elapsedMs)}ms > " +
+ log"${MDC(MIN_TIME, minTimeToSpeculation)}ms")
}
}
foundTasks
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
index 9ddc63b27537..d359b65caa93 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
@@ -33,7 +33,8 @@ import org.apache.spark.deploy.security.HadoopDelegationTokenManager
import org.apache.spark.errors.SparkCoreErrors
import org.apache.spark.executor.ExecutorLogUrlHandler
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.ERROR
+import org.apache.spark.internal.LogKeys
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.Network._
import org.apache.spark.resource.ResourceProfile
@@ -181,8 +182,9 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
makeOffers(executorId)
case None =>
// Ignoring the update since we don't know about the executor.
- logWarning(s"Ignored task status update ($taskId state $state) " +
- s"from unknown executor with ID $executorId")
+ logWarning(log"Ignored task status update (${MDC(TASK_ID, taskId)} " +
+ log"state ${MDC(TASK_STATE, state)}) " +
+ log"from unknown executor with ID ${MDC(LogKeys.EXECUTOR_ID, executorId)}")
}
}
@@ -199,7 +201,8 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
KillTask(taskId, executorId, interruptThread, reason))
case None =>
// Ignoring the task kill since the executor is not registered.
- logWarning(s"Attempted to kill task $taskId for unknown executor $executorId.")
+ logWarning(log"Attempted to kill task ${MDC(TASK_ID, taskId)} " +
+ log"for unknown executor ${MDC(LogKeys.EXECUTOR_ID, executorId)}.")
}
case KillExecutorsOnHost(host) =>
@@ -340,7 +343,8 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
// Do not change this code without running the K8s integration suites
case ExecutorDecommissioning(executorId) =>
- logWarning(s"Received executor $executorId decommissioned message")
+ logWarning(log"Received executor ${MDC(LogKeys.EXECUTOR_ID, executorId)} " +
+ log"decommissioned message")
context.reply(
decommissionExecutor(
executorId,
@@ -952,7 +956,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
val response = withLock {
val (knownExecutors, unknownExecutors) = executorIds.partition(executorDataMap.contains)
unknownExecutors.foreach { id =>
- logWarning(s"Executor to kill $id does not exist!")
+ logWarning(log"Executor to kill ${MDC(LogKeys.EXECUTOR_ID, id)} does not exist!")
}
// If an executor is already pending to be removed, do not kill it again (SPARK-9795)
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala
index 144ea1b31ca3..8f15dec6739a 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/StandaloneSchedulerBackend.scala
@@ -349,8 +349,8 @@ private[spark] class StandaloneSchedulerBackend(
_executorRemoveDelay, TimeUnit.MILLISECONDS)
} catch {
case _: RejectedExecutionException if stopping.get() =>
- logWarning(
- "Skipping onDisconnected RemoveExecutor call because the scheduler is stopping")
+ logWarning("Skipping onDisconnected RemoveExecutor call " +
+ "because the scheduler is stopping")
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/security/CryptoStreamUtils.scala b/core/src/main/scala/org/apache/spark/security/CryptoStreamUtils.scala
index 409223132a62..1ee46d51ce70 100644
--- a/core/src/main/scala/org/apache/spark/security/CryptoStreamUtils.scala
+++ b/core/src/main/scala/org/apache/spark/security/CryptoStreamUtils.scala
@@ -31,7 +31,8 @@ import org.apache.commons.crypto.random._
import org.apache.commons.crypto.stream._
import org.apache.spark.SparkConf
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.network.util.{CryptoUtils, JavaUtils}
@@ -131,8 +132,8 @@ private[spark] object CryptoStreamUtils extends Logging {
val initialIVFinish = System.nanoTime()
val initialIVTime = TimeUnit.NANOSECONDS.toMillis(initialIVFinish - initialIVStart)
if (initialIVTime > 2000) {
- logWarning(s"It costs ${initialIVTime} milliseconds to create the Initialization Vector " +
- s"used by CryptoStream")
+ logWarning(log"It costs ${MDC(TIME_UNITS, initialIVTime)} milliseconds " +
+ log"to create the Initialization Vector used by CryptoStream")
}
iv
}
diff --git a/core/src/main/scala/org/apache/spark/shuffle/IndexShuffleBlockResolver.scala b/core/src/main/scala/org/apache/spark/shuffle/IndexShuffleBlockResolver.scala
index a1a204dfc01b..d0a202cb7951 100644
--- a/core/src/main/scala/org/apache/spark/shuffle/IndexShuffleBlockResolver.scala
+++ b/core/src/main/scala/org/apache/spark/shuffle/IndexShuffleBlockResolver.scala
@@ -29,7 +29,7 @@ import com.google.common.cache.CacheBuilder
import org.apache.spark.{SecurityManager, SparkConf, SparkEnv, SparkException}
import org.apache.spark.errors.SparkCoreErrors
import org.apache.spark.internal.{config, Logging, MDC}
-import org.apache.spark.internal.LogKeys.PATH
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.io.NioBufferedFileInputStream
import org.apache.spark.network.buffer.{FileSegmentManagedBuffer, ManagedBuffer}
import org.apache.spark.network.client.StreamCallbackWithID
@@ -172,17 +172,17 @@ private[spark] class IndexShuffleBlockResolver(
def removeDataByMap(shuffleId: Int, mapId: Long): Unit = {
var file = getDataFile(shuffleId, mapId)
if (file.exists() && !file.delete()) {
- logWarning(s"Error deleting data ${file.getPath()}")
+ logWarning(log"Error deleting data ${MDC(PATH, file.getPath())}")
}
file = getIndexFile(shuffleId, mapId)
if (file.exists() && !file.delete()) {
- logWarning(s"Error deleting index ${file.getPath()}")
+ logWarning(log"Error deleting index ${MDC(PATH, file.getPath())}")
}
file = getChecksumFile(shuffleId, mapId, conf.get(config.SHUFFLE_CHECKSUM_ALGORITHM))
if (file.exists() && !file.delete()) {
- logWarning(s"Error deleting checksum ${file.getPath()}")
+ logWarning(log"Error deleting checksum ${MDC(PATH, file.getPath())}")
}
}
@@ -290,7 +290,7 @@ private[spark] class IndexShuffleBlockResolver(
override def onFailure(streamId: String, cause: Throwable): Unit = {
// the framework handles the connection itself, we just need to do local cleanup
- logWarning(s"Error while uploading $blockId", cause)
+ logWarning(log"Error while uploading ${MDC(BLOCK_ID, blockId)}", cause)
channel.close()
fileTmp.delete()
}
@@ -328,8 +328,9 @@ private[spark] class IndexShuffleBlockResolver(
}
} catch {
case _: Exception => // If we can't load the blocks ignore them.
- logWarning(s"Failed to resolve shuffle block ${shuffleBlockInfo}. " +
- "This is expected to occur if a block is removed after decommissioning has started.")
+ logWarning(log"Failed to resolve shuffle block " +
+ log"${MDC(SHUFFLE_BLOCK_INFO, shuffleBlockInfo)}. " +
+ log"This is expected to occur if a block is removed after decommissioning has started.")
List.empty[(BlockId, ManagedBuffer)]
}
}
@@ -485,7 +486,8 @@ private[spark] class IndexShuffleBlockResolver(
if (propagateError) {
throw SparkCoreErrors.failedRenameTempFileError(tmpFile, targetFile)
} else {
- logWarning(s"fail to rename file $tmpFile to $targetFile")
+ logWarning(log"fail to rename file ${MDC(TEMP_FILE, tmpFile)} " +
+ log"to ${MDC(TARGET_PATH, targetFile)}")
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala b/core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala
index bd03934cada4..4e3191e44fbd 100644
--- a/core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala
+++ b/core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala
@@ -27,7 +27,8 @@ import scala.util.control.NonFatal
import org.apache.spark.{SecurityManager, ShuffleDependency, SparkConf, SparkContext, SparkEnv}
import org.apache.spark.annotation.Since
import org.apache.spark.executor.{CoarseGrainedExecutorBackend, ExecutorBackend}
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.launcher.SparkLauncher
import org.apache.spark.network.buffer.{FileSegmentManagedBuffer, ManagedBuffer, NioManagedBuffer}
@@ -248,7 +249,8 @@ private[spark] class ShuffleBlockPusher(conf: SparkConf) extends Logging {
if (!errorHandler.shouldLogError(exception)) {
logTrace(s"Pushing block $blockId to $address failed.", exception)
} else {
- logWarning(s"Pushing block $blockId to $address failed.", exception)
+ logWarning(log"Pushing block ${MDC(BLOCK_ID, blockId)} " +
+ log"to ${MDC(HOST_PORT, address)} failed.", exception)
}
handleResult(PushResult(blockId, exception))
}
@@ -329,9 +331,9 @@ private[spark] class ShuffleBlockPusher(conf: SparkConf) extends Logging {
unreachableBlockMgrs.add(address)
removed += pushRequests.dequeueAll(req => req.address == address).length
removed += deferredPushRequests.remove(address).map(_.length).getOrElse(0)
- logWarning(s"Received a ConnectException from $address. " +
- s"Dropping $removed push-requests and " +
- s"not pushing any more blocks to this address.")
+ logWarning(log"Received a ConnectException from ${MDC(HOST_PORT, address)}. " +
+ log"Dropping ${MDC(NUM_REQUESTS, removed)} push-requests and " +
+ log"not pushing any more blocks to this address.")
}
}
if (pushResult.failure != null && !errorHandler.shouldRetryError(pushResult.failure)) {
@@ -360,7 +362,8 @@ private[spark] class ShuffleBlockPusher(conf: SparkConf) extends Logging {
case Some(cb: CoarseGrainedExecutorBackend) =>
cb.notifyDriverAboutPushCompletion(shuffleId, shuffleMergeId, mapIndex)
case Some(eb: ExecutorBackend) =>
- logWarning(s"Currently $eb doesn't support push-based shuffle")
+ logWarning(log"Currently ${MDC(EXECUTOR_BACKEND, eb)} " +
+ log"doesn't support push-based shuffle")
case None =>
}
pushCompletionNotified = true
diff --git a/core/src/main/scala/org/apache/spark/status/KVUtils.scala b/core/src/main/scala/org/apache/spark/status/KVUtils.scala
index 821686803f46..e334626413dc 100644
--- a/core/src/main/scala/org/apache/spark/status/KVUtils.scala
+++ b/core/src/main/scala/org/apache/spark/status/KVUtils.scala
@@ -31,7 +31,8 @@ import org.rocksdb.RocksDBException
import org.apache.spark.SparkConf
import org.apache.spark.deploy.history.{FsHistoryProvider, FsHistoryProviderMetadata}
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.History
import org.apache.spark.internal.config.History.HYBRID_STORE_DISK_BACKEND
import org.apache.spark.internal.config.History.HybridStoreDiskBackend
@@ -154,7 +155,7 @@ private[spark] object KVUtils extends Logging {
open(dbPath, metadata, conf, live)
case dbExc @ (_: NativeDB.DBException | _: RocksDBException) =>
// Get rid of the corrupted data and re-create it.
- logWarning(s"Failed to load disk store $dbPath :", dbExc)
+ logWarning(log"Failed to load disk store ${MDC(PATH, dbPath)} :", dbExc)
Utils.deleteRecursively(dbPath)
open(dbPath, metadata, conf, live)
}
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
index 740031d70752..26e2acf4392c 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
@@ -42,7 +42,7 @@ import org.apache.spark._
import org.apache.spark.errors.SparkCoreErrors
import org.apache.spark.executor.DataReadMethod
import org.apache.spark.internal.{config, Logging, MDC}
-import org.apache.spark.internal.LogKeys.{BLOCK_ID, COUNT, SLEEP_TIME}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.{Network, RDD_CACHE_VISIBILITY_TRACKING_ENABLED, Tests}
import org.apache.spark.memory.{MemoryManager, MemoryMode}
import org.apache.spark.metrics.source.Source
@@ -423,7 +423,7 @@ private[spark] class BlockManager(
saveSerializedValuesToMemoryStore(readToByteBuffer())
}
if (!putSucceeded && level.useDisk) {
- logWarning(s"Persisting block $blockId to disk instead.")
+ logWarning(log"Persisting block ${MDC(BLOCK_ID, blockId)} to disk instead.")
saveToDiskStore()
}
} else if (level.useDisk) {
@@ -1214,14 +1214,16 @@ private[spark] class BlockManager(
// Give up trying anymore locations. Either we've tried all of the original locations,
// or we've refreshed the list of locations from the master, and have still
// hit failures after trying locations from the refreshed list.
- logWarning(s"Failed to fetch remote block $blockId " +
- s"from [${locations.mkString(", ")}] after $totalFailureCount fetch failures. " +
- s"Most recent failure cause:", e)
+ logWarning(log"Failed to fetch remote block ${MDC(BLOCK_ID, blockId)} " +
+ log"from [${MDC(BLOCK_MANAGER_IDS, locations.mkString(", "))}] " +
+ log"after ${MDC(NUM_FAILURES, totalFailureCount)} fetch failures. " +
+ log"Most recent failure cause:", e)
return None
}
- logWarning(s"Failed to fetch remote block $blockId " +
- s"from $loc (failed attempt $runningFailureCount)", e)
+ logWarning(log"Failed to fetch remote block ${MDC(BLOCK_ID, blockId)} " +
+ log"from ${MDC(BLOCK_MANAGER_ID, loc)} " +
+ log"(failed attempt ${MDC(NUM_FAILURES, runningFailureCount)}", e)
// If there is a large number of executors then locations list can contain a
// large number of stale entries causing a large number of retries that may
@@ -1335,7 +1337,8 @@ private[spark] class BlockManager(
// SPARK-27666. When a task completes, Spark automatically releases all the blocks locked
// by this task. We should not release any locks for a task that is already completed.
if (taskContext.isDefined && taskContext.get.isCompleted()) {
- logWarning(s"Task ${taskAttemptId.get} already completed, not releasing lock for $blockId")
+ logWarning(log"Task ${MDC(TASK_ATTEMPT_ID, taskAttemptId.get)} " +
+ log"already completed, not releasing lock for ${MDC(BLOCK_ID, blockId)}")
} else {
blockInfoManager.unlock(blockId, taskAttemptId)
}
@@ -1544,7 +1547,8 @@ private[spark] class BlockManager(
if (blockInfoManager.lockNewBlockForWriting(blockId, newInfo, keepReadLock)) {
newInfo
} else {
- logWarning(s"Block $blockId already exists on this machine; not re-adding it")
+ logWarning(log"Block ${MDC(BLOCK_ID, blockId)} " +
+ log"already exists on this machine; not re-adding it")
return None
}
}
@@ -1562,7 +1566,7 @@ private[spark] class BlockManager(
blockInfoManager.unlock(blockId)
}
} else {
- logWarning(s"Putting block $blockId failed")
+ logWarning(log"Putting block ${MDC(BLOCK_ID, blockId)} failed")
removeBlockInternal(blockId, tellMaster = false)
}
res
@@ -1570,7 +1574,8 @@ private[spark] class BlockManager(
// Since removeBlockInternal may throw exception,
// we should print exception first to show root cause.
case NonFatal(e) =>
- logWarning(s"Putting block $blockId failed due to exception $e.")
+ logWarning(log"Putting block ${MDC(BLOCK_ID, blockId)} " +
+ log"failed due to exception ${MDC(ERROR, e)}.")
throw e
} finally {
// This cleanup is performed in a finally block rather than a `catch` to avoid having to
@@ -1631,7 +1636,7 @@ private[spark] class BlockManager(
case Left(iter) =>
// Not enough space to unroll this block; drop to disk if applicable
if (level.useDisk) {
- logWarning(s"Persisting block $blockId to disk instead.")
+ logWarning(log"Persisting block ${MDC(BLOCK_ID, blockId)} to disk instead.")
diskStore.put(blockId) { channel =>
val out = Channels.newOutputStream(channel)
serializerManager.dataSerializeStream(blockId, out, iter)(classTag)
@@ -1648,7 +1653,7 @@ private[spark] class BlockManager(
case Left(partiallySerializedValues) =>
// Not enough space to unroll this block; drop to disk if applicable
if (level.useDisk) {
- logWarning(s"Persisting block $blockId to disk instead.")
+ logWarning(log"Persisting block ${MDC(BLOCK_ID, blockId)} to disk instead.")
diskStore.put(blockId) { channel =>
val out = Channels.newOutputStream(channel)
partiallySerializedValues.finishWritingToStream(out)
@@ -1904,7 +1909,9 @@ private[spark] class BlockManager(
throw e
// Everything else we may retry
case NonFatal(e) =>
- logWarning(s"Failed to replicate $blockId to $peer, failure #$numFailures", e)
+ logWarning(log"Failed to replicate ${MDC(BLOCK_ID, blockId)} " +
+ log"to ${MDC(PEER, peer)}, " +
+ log"failure #${MDC(NUM_FAILURES, numFailures)}", e)
peersFailedToReplicateTo += peer
// we have a failed replication, so we get the list of peers again
// we don't want peers we have already replicated to and the ones that
@@ -1925,8 +1932,9 @@ private[spark] class BlockManager(
logDebug(s"Replicating $blockId of ${data.size} bytes to " +
s"${peersReplicatedTo.size} peer(s) took ${(System.nanoTime - startTime) / 1e6} ms")
if (peersReplicatedTo.size < numPeersToReplicateTo) {
- logWarning(s"Block $blockId replicated to only " +
- s"${peersReplicatedTo.size} peer(s) instead of $numPeersToReplicateTo peers")
+ logWarning(log"Block ${MDC(BLOCK_ID, blockId)} replicated to only " +
+ log"${MDC(NUM_PEERS_REPLICATED_TO, peersReplicatedTo.size)} peer(s) " +
+ log"instead of ${MDC(NUM_PEERS_TO_REPLICATE_TO, numPeersToReplicateTo)} peers")
return false
}
@@ -1999,7 +2007,8 @@ private[spark] class BlockManager(
if (blockIsRemoved) {
blockIsUpdated = true
} else {
- logWarning(s"Block $blockId could not be dropped from memory as it does not exist")
+ logWarning(log"Block ${MDC(BLOCK_ID, blockId)} " +
+ log"could not be dropped from memory as it does not exist")
}
val status = getCurrentBlockStatus(blockId, info)
@@ -2083,7 +2092,8 @@ private[spark] class BlockManager(
blockInfoManager.lockForWriting(blockId) match {
case None =>
// The block has already been removed; do nothing.
- logWarning(s"Asked to remove block $blockId, which does not exist")
+ logWarning(log"Asked to remove block ${MDC(BLOCK_ID, blockId)}, " +
+ log"which does not exist")
case Some(info) =>
removeBlockInternal(blockId, tellMaster = tellMaster && info.tellMaster)
addUpdatedBlockStatusToTaskMetrics(blockId, BlockStatus.empty)
@@ -2106,7 +2116,8 @@ private[spark] class BlockManager(
val removedFromMemory = memoryStore.remove(blockId)
val removedFromDisk = diskStore.remove(blockId)
if (!removedFromMemory && !removedFromDisk) {
- logWarning(s"Block $blockId could not be removed as it was not found on disk or in memory")
+ logWarning(log"Block ${MDC(BLOCK_ID, blockId)} " +
+ log"could not be removed as it was not found on disk or in memory")
}
blockInfoManager.removeBlock(blockId)
@@ -2118,7 +2129,7 @@ private[spark] class BlockManager(
}
} finally {
if (!hasRemoveBlock) {
- logWarning(s"Block $blockId was not removed normally.")
+ logWarning(log"Block ${MDC(BLOCK_ID, blockId)} was not removed normally.")
blockInfoManager.removeBlock(blockId)
}
}
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala
index ad4533166316..fc98fbf6e18b 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerDecommissioner.scala
@@ -27,7 +27,7 @@ import scala.util.control.NonFatal
import org.apache.spark._
import org.apache.spark.errors.SparkCoreErrors
import org.apache.spark.internal.{config, Logging, MDC}
-import org.apache.spark.internal.LogKeys.SHUFFLE_BLOCK_INFO
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.shuffle.ShuffleBlockInfo
import org.apache.spark.storage.BlockManagerMessages.ReplicateBlock
import org.apache.spark.util.{ThreadUtils, Utils}
@@ -78,8 +78,9 @@ private[storage] class BlockManagerDecommissioner(
// The block needs to retry so we should not mark it as finished
shufflesToMigrate.add((shuffleBlock, failureNum))
} else {
- logWarning(s"Give up migrating $shuffleBlock since it's been " +
- s"failed for $maxReplicationFailuresForDecommission times")
+ logWarning(log"Give up migrating ${MDC(SHUFFLE_BLOCK_INFO, shuffleBlock)} " +
+ log"since it's been failed for " +
+ log"${MDC(NUM_FAILURES, maxReplicationFailuresForDecommission)} times")
false
}
}
@@ -140,7 +141,8 @@ private[storage] class BlockManagerDecommissioner(
// could also happen with manually managed shuffles or a GC event on the
// driver a no longer referenced RDD with shuffle files.
if (bm.migratableResolver.getMigrationBlocks(shuffleBlockInfo).size < blocks.size) {
- logWarning(s"Skipping block $shuffleBlockInfo, block deleted.")
+ logWarning(log"Skipping block ${MDC(SHUFFLE_BLOCK_INFO, shuffleBlockInfo)}, " +
+ log"block deleted.")
} else if (fallbackStorage.isDefined
// Confirm peer is not the fallback BM ID because fallbackStorage would already
// have been used in the try-block above so there's no point trying again
@@ -165,7 +167,7 @@ private[storage] class BlockManagerDecommissioner(
if (keepRunning) {
numMigratedShuffles.incrementAndGet()
} else {
- logWarning(s"Stop migrating shuffle blocks to $peer")
+ logWarning(log"Stop migrating shuffle blocks to ${MDC(PEER, peer)}")
val newRetryCount = if (isTargetDecommissioned) {
retryCount
@@ -351,7 +353,7 @@ private[storage] class BlockManagerDecommissioner(
logInfo(s"Need to replicate ${replicateBlocksInfo.size} RDD blocks " +
"for block manager decommissioning")
} else {
- logWarning(s"Asked to decommission RDD cache blocks, but no blocks to migrate")
+ logWarning("Asked to decommission RDD cache blocks, but no blocks to migrate")
return false
}
@@ -362,8 +364,8 @@ private[storage] class BlockManagerDecommissioner(
(replicateBlock.blockId, replicatedSuccessfully)
}.filterNot(_._2).map(_._1)
if (blocksFailedReplication.nonEmpty) {
- logWarning("Blocks failed replication in cache decommissioning " +
- s"process: ${blocksFailedReplication.mkString(",")}")
+ logWarning(log"Blocks failed replication in cache decommissioning " +
+ log"process: ${MDC(BLOCK_IDS, blocksFailedReplication.mkString(","))}")
return true
}
false
@@ -380,7 +382,7 @@ private[storage] class BlockManagerDecommissioner(
bm.removeBlock(blockToReplicate.blockId)
logInfo(s"Block ${blockToReplicate.blockId} removed")
} else {
- logWarning(s"Failed to migrate block ${blockToReplicate.blockId}")
+ logWarning(log"Failed to migrate block ${MDC(BLOCK_ID, blockToReplicate.blockId)}")
}
replicatedSuccessfully
}
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
index 2b961317e01d..95af44deef93 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
@@ -23,7 +23,8 @@ import scala.concurrent.Future
import org.apache.spark.SparkConf
import org.apache.spark.errors.SparkCoreErrors
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.rpc.RpcEndpointRef
import org.apache.spark.storage.BlockManagerMessages._
import org.apache.spark.util.{RpcUtils, ThreadUtils}
@@ -189,7 +190,8 @@ class BlockManagerMaster(
def removeRdd(rddId: Int, blocking: Boolean): Unit = {
val future = driverEndpoint.askSync[Future[Seq[Int]]](RemoveRdd(rddId))
future.failed.foreach(e =>
- logWarning(s"Failed to remove RDD $rddId - ${e.getMessage}", e)
+ logWarning(log"Failed to remove RDD ${MDC(RDD_ID, rddId)} - " +
+ log"${MDC(ERROR, e.getMessage)}", e)
)(ThreadUtils.sameThread)
if (blocking) {
// the underlying Futures will timeout anyway, so it's safe to use infinite timeout here
@@ -201,7 +203,8 @@ class BlockManagerMaster(
def removeShuffle(shuffleId: Int, blocking: Boolean): Unit = {
val future = driverEndpoint.askSync[Future[Seq[Boolean]]](RemoveShuffle(shuffleId))
future.failed.foreach(e =>
- logWarning(s"Failed to remove shuffle $shuffleId - ${e.getMessage}", e)
+ logWarning(log"Failed to remove shuffle ${MDC(SHUFFLE_ID, shuffleId)} - " +
+ log"${MDC(ERROR, e.getMessage)}", e)
)(ThreadUtils.sameThread)
if (blocking) {
// the underlying Futures will timeout anyway, so it's safe to use infinite timeout here
@@ -214,8 +217,9 @@ class BlockManagerMaster(
val future = driverEndpoint.askSync[Future[Seq[Int]]](
RemoveBroadcast(broadcastId, removeFromMaster))
future.failed.foreach(e =>
- logWarning(s"Failed to remove broadcast $broadcastId" +
- s" with removeFromMaster = $removeFromMaster - ${e.getMessage}", e)
+ logWarning(log"Failed to remove broadcast ${MDC(BROADCAST_ID, broadcastId)}" +
+ log" with removeFromMaster = ${MDC(REMOVE_FROM_MASTER, removeFromMaster)} - " +
+ log"${MDC(ERROR, e.getMessage)}", e)
)(ThreadUtils.sameThread)
if (blocking) {
// the underlying Futures will timeout anyway, so it's safe to use infinite timeout here
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
index cf3e755fc4f9..be7082807182 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
@@ -32,7 +32,7 @@ import com.google.common.cache.CacheBuilder
import org.apache.spark.{MapOutputTrackerMaster, SparkConf, SparkContext, SparkEnv}
import org.apache.spark.annotation.DeveloperApi
import org.apache.spark.internal.{config, Logging, MDC}
-import org.apache.spark.internal.LogKeys.{BLOCK_MANAGER_ID, EXECUTOR_ID, OLD_BLOCK_MANAGER_ID}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.RDD_CACHE_VISIBILITY_TRACKING_ENABLED
import org.apache.spark.network.shuffle.{ExternalBlockStoreClient, RemoteBlockPushResolver}
import org.apache.spark.rpc.{IsolatedThreadSafeRpcEndpoint, RpcCallContext, RpcEndpointRef, RpcEnv}
@@ -314,8 +314,9 @@ class BlockManagerMasterEndpoint(
defaultValue: T): PartialFunction[Throwable, T] = {
case e: IOException =>
if (!SparkContext.getActive.map(_.isStopped).getOrElse(true)) {
- logWarning(s"Error trying to remove $blockType $blockId" +
- s" from block manager $bmId", e)
+ logWarning(log"Error trying to remove ${MDC(BLOCK_TYPE, blockType)} " +
+ log"${MDC(BLOCK_ID, blockId)}" +
+ log" from block manager ${MDC(BLOCK_MANAGER_ID, bmId)}", e)
}
defaultValue
@@ -333,8 +334,9 @@ class BlockManagerMasterEndpoint(
false
}
if (!isAlive) {
- logWarning(s"Error trying to remove $blockType $blockId. " +
- s"The executor $executorId may have been lost.", t)
+ logWarning(log"Error trying to remove ${MDC(BLOCK_TYPE, blockType)} " +
+ log"${MDC(BLOCK_ID, blockId)}. " +
+ log"The executor ${MDC(EXECUTOR_ID, executorId)} may have been lost.", t)
defaultValue
} else {
throw t
@@ -516,7 +518,7 @@ class BlockManagerMasterEndpoint(
// etc.) as replication doesn't make much sense in that context.
if (locations.isEmpty) {
blockLocations.remove(blockId)
- logWarning(s"No more replicas available for $blockId !")
+ logWarning(log"No more replicas available for ${MDC(BLOCK_ID, blockId)}!")
} else if (proactivelyReplicate && (blockId.isRDD || blockId.isInstanceOf[TestBlockId])) {
// As a heuristic, assume single executor failure to find out the number of replicas that
// existed before failure
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockReplicationPolicy.scala b/core/src/main/scala/org/apache/spark/storage/BlockReplicationPolicy.scala
index 893b5605414e..5186cbfa217c 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockReplicationPolicy.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockReplicationPolicy.scala
@@ -21,7 +21,8 @@ import scala.collection.mutable
import scala.util.Random
import org.apache.spark.annotation.DeveloperApi
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
/**
* ::DeveloperApi::
@@ -120,7 +121,8 @@ class RandomBlockReplicationPolicy
BlockReplicationUtils.getRandomSample(peers, numReplicas, random)
} else {
if (peers.size < numReplicas) {
- logWarning(s"Expecting ${numReplicas} replicas with only ${peers.size} peer/s.")
+ logWarning(log"Expecting ${MDC(NUM_REPLICAS, numReplicas)} " +
+ log"replicas with only ${MDC(NUM_PEERS, peers.size)} peer/s.")
}
random.shuffle(peers).toList
}
diff --git a/core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala b/core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala
index ad60b66b4f7a..6a6446437d82 100644
--- a/core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala
+++ b/core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala
@@ -325,8 +325,8 @@ private[spark] class DiskBlockManager(
logDebug(s"Created directory at ${dirToCreate.getAbsolutePath} with permission 770")
} catch {
case e: SecurityException =>
- logWarning(s"Failed to create directory ${dirToCreate.getAbsolutePath} " +
- s"with permission 770", e)
+ logWarning(log"Failed to create directory ${MDC(PATH, dirToCreate.getAbsolutePath)} " +
+ log"with permission 770", e)
created = null;
}
}
diff --git a/core/src/main/scala/org/apache/spark/storage/DiskBlockObjectWriter.scala b/core/src/main/scala/org/apache/spark/storage/DiskBlockObjectWriter.scala
index d6707419d2fb..e02357f46c5d 100644
--- a/core/src/main/scala/org/apache/spark/storage/DiskBlockObjectWriter.scala
+++ b/core/src/main/scala/org/apache/spark/storage/DiskBlockObjectWriter.scala
@@ -25,7 +25,7 @@ import java.util.zip.Checksum
import org.apache.spark.SparkException
import org.apache.spark.errors.SparkCoreErrors
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.{ERROR, PATH}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.io.MutableCheckedOutputStream
import org.apache.spark.serializer.{SerializationStream, SerializerInstance, SerializerManager}
import org.apache.spark.shuffle.ShuffleWriteMetricsReporter
@@ -297,7 +297,7 @@ private[spark] class DiskBlockObjectWriter(
}
} {
if (!Files.deleteIfExists(file.toPath)) {
- logWarning(s"Error deleting $file")
+ logWarning(log"Error deleting ${MDC(FILE_NAME, file)}")
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/storage/DiskStore.scala b/core/src/main/scala/org/apache/spark/storage/DiskStore.scala
index 442b74ba9a23..1498b224b0c9 100644
--- a/core/src/main/scala/org/apache/spark/storage/DiskStore.scala
+++ b/core/src/main/scala/org/apache/spark/storage/DiskStore.scala
@@ -30,7 +30,8 @@ import io.netty.channel.DefaultFileRegion
import org.apache.commons.io.FileUtils
import org.apache.spark.{SecurityManager, SparkConf, SparkException}
-import org.apache.spark.internal.{config, Logging}
+import org.apache.spark.internal.{config, Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.network.buffer.ManagedBuffer
import org.apache.spark.network.util.{AbstractFileRegion, JavaUtils}
import org.apache.spark.security.CryptoStreamUtils
@@ -62,7 +63,7 @@ private[spark] class DiskStore(
*/
def put(blockId: BlockId)(writeFunc: WritableByteChannel => Unit): Unit = {
if (contains(blockId)) {
- logWarning(s"Block $blockId is already present in the disk store")
+ logWarning(log"Block ${MDC(BLOCK_ID, blockId)} is already present in the disk store")
try {
diskManager.getFile(blockId).delete()
} catch {
@@ -133,7 +134,7 @@ private[spark] class DiskStore(
if (file.exists()) {
val ret = file.delete()
if (!ret) {
- logWarning(s"Error deleting ${file.getPath()}")
+ logWarning(log"Error deleting ${MDC(PATH, file.getPath())}")
}
ret
} else {
diff --git a/core/src/main/scala/org/apache/spark/storage/FallbackStorage.scala b/core/src/main/scala/org/apache/spark/storage/FallbackStorage.scala
index 161120393490..00297eca29a2 100644
--- a/core/src/main/scala/org/apache/spark/storage/FallbackStorage.scala
+++ b/core/src/main/scala/org/apache/spark/storage/FallbackStorage.scala
@@ -28,7 +28,8 @@ import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.deploy.SparkHadoopUtil
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.{STORAGE_DECOMMISSION_FALLBACK_STORAGE_CLEANUP, STORAGE_DECOMMISSION_FALLBACK_STORAGE_PATH}
import org.apache.spark.network.buffer.{ManagedBuffer, NioManagedBuffer}
import org.apache.spark.network.util.JavaUtils
@@ -84,7 +85,7 @@ private[storage] class FallbackStorage(conf: SparkConf) extends Logging {
}
}
case r =>
- logWarning(s"Unsupported Resolver: ${r.getClass.getName}")
+ logWarning(log"Unsupported Resolver: ${MDC(CLASS_NAME, r.getClass.getName)}")
}
}
@@ -141,7 +142,7 @@ private[spark] object FallbackStorage extends Logging {
logInfo(s"Succeed to clean up: $fallbackUri")
} else {
// Clean-up can fail due to the permission issues.
- logWarning(s"Failed to clean up: $fallbackUri")
+ logWarning(log"Failed to clean up: ${MDC(URI, fallbackUri)}")
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/storage/PushBasedFetchHelper.scala b/core/src/main/scala/org/apache/spark/storage/PushBasedFetchHelper.scala
index 52056594e792..059bb5271410 100644
--- a/core/src/main/scala/org/apache/spark/storage/PushBasedFetchHelper.scala
+++ b/core/src/main/scala/org/apache/spark/storage/PushBasedFetchHelper.scala
@@ -29,7 +29,7 @@ import org.roaringbitmap.RoaringBitmap
import org.apache.spark.MapOutputTracker
import org.apache.spark.MapOutputTracker.SHUFFLE_PUSH_MAP_ID
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.{HOST, PORT, REDUCE_ID, SHUFFLE_ID, SHUFFLE_MERGE_ID}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.network.shuffle.{BlockStoreClient, MergedBlockMeta, MergedBlocksMetaListener}
import org.apache.spark.shuffle.ShuffleReadMetricsReporter
import org.apache.spark.storage.BlockManagerId.SHUFFLE_MERGER_IDENTIFIER
@@ -246,8 +246,9 @@ private class PushBasedFetchHelper(
case Failure(throwable) =>
// If we see an exception with getting the local dirs for push-merged-local blocks,
// we fallback to fetch the original blocks. We do not report block fetch failure.
- logWarning(s"Error while fetching the merged dirs for push-merged-local " +
- s"blocks: ${pushMergedLocalBlocks.mkString(", ")}. Fetch the original blocks instead",
+ logWarning(log"Error while fetching the merged dirs for push-merged-local " +
+ log"blocks: ${MDC(BLOCK_IDS, pushMergedLocalBlocks.mkString(", "))}. " +
+ log"Fetch the original blocks instead",
throwable)
pushMergedLocalBlocks.foreach {
blockId =>
@@ -280,8 +281,8 @@ private class PushBasedFetchHelper(
// If we see an exception with reading a push-merged-local meta, we fallback to
// fetch the original blocks. We do not report block fetch failure
// and will continue with the remaining local block read.
- logWarning(s"Error occurred while fetching push-merged-local meta, " +
- s"prepare to fetch the original blocks", e)
+ logWarning(log"Error occurred while fetching push-merged-local meta, " +
+ log"prepare to fetch the original blocks", e)
iterator.addToResultsQueue(
FallbackOnPushMergedFailureResult(blockId, blockManagerId, 0, isNetworkReqDone = false))
}
@@ -315,7 +316,8 @@ private class PushBasedFetchHelper(
blockId: BlockId,
address: BlockManagerId): Unit = {
assert(blockId.isInstanceOf[ShuffleMergedBlockId] || blockId.isInstanceOf[ShuffleBlockChunkId])
- logWarning(s"Falling back to fetch the original blocks for push-merged block $blockId")
+ logWarning(log"Falling back to fetch the original blocks for push-merged block " +
+ log"${MDC(BLOCK_ID, blockId)}")
shuffleMetrics.incMergedFetchFallbackCount(1)
// Increase the blocks processed since we will process another block in the next iteration of
// the while loop in ShuffleBlockFetcherIterator.next().
diff --git a/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala b/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala
index 96508649522c..b89d24f4c6dd 100644
--- a/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala
+++ b/core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala
@@ -37,7 +37,7 @@ import org.apache.spark.{MapOutputTracker, SparkException, TaskContext}
import org.apache.spark.MapOutputTracker.SHUFFLE_PUSH_MAP_ID
import org.apache.spark.errors.SparkCoreErrors
import org.apache.spark.internal.{Logging, MDC}
-import org.apache.spark.internal.LogKeys.{BLOCK_ID, ERROR, HOST, MAX_ATTEMPTS, PORT}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.network.buffer.{FileSegmentManagedBuffer, ManagedBuffer}
import org.apache.spark.network.shuffle._
import org.apache.spark.network.shuffle.checksum.{Cause, ShuffleChecksumHelper}
@@ -249,7 +249,7 @@ final class ShuffleBlockFetcherIterator(
}
shuffleFilesSet.foreach { file =>
if (!file.delete()) {
- logWarning("Failed to cleanup shuffle fetch temp file " + file.path())
+ logWarning(log"Failed to cleanup shuffle fetch temp file ${MDC(PATH, file.path())}")
}
}
}
@@ -846,8 +846,10 @@ final class ShuffleBlockFetcherIterator(
// uses are shared by the UnsafeShuffleWriter (both writers use DiskBlockObjectWriter
// which returns a zero-size from commitAndGet() in case no records were written
// since the last call.
- val msg = s"Received a zero-size buffer for block $blockId from $address " +
- s"(expectedApproxSize = $size, isNetworkReqDone=$isNetworkReqDone)"
+ val msg = log"Received a zero-size buffer for block ${MDC(BLOCK_ID, blockId)} " +
+ log"from ${MDC(URI, address)} " +
+ log"(expectedApproxSize = ${MDC(NUM_BYTES, size)}, " +
+ log"isNetworkReqDone=${MDC(IS_NETWORK_REQUEST_DONE, isNetworkReqDone)})"
if (blockId.isShuffleChunk) {
// Zero-size block may come from nodes with hardware failures, For shuffle chunks,
// the original shuffle blocks that belong to that zero-size shuffle chunk is
@@ -859,7 +861,7 @@ final class ShuffleBlockFetcherIterator(
result = null
null
} else {
- throwFetchFailedException(blockId, mapIndex, address, new IOException(msg))
+ throwFetchFailedException(blockId, mapIndex, address, new IOException(msg.message))
}
} else {
try {
@@ -945,7 +947,8 @@ final class ShuffleBlockFetcherIterator(
}
} else {
// It's the first time this block is detected corrupted
- logWarning(s"got an corrupted block $blockId from $address, fetch again", e)
+ logWarning(log"got an corrupted block ${MDC(BLOCK_ID, blockId)} " +
+ log"from ${MDC(URI, address)}, fetch again", e)
corruptedBlocks += blockId
fetchRequests += FetchRequest(
address, Array(FetchBlockInfo(blockId, size, mapIndex)))
@@ -1033,8 +1036,8 @@ final class ShuffleBlockFetcherIterator(
// If we see an exception with reading push-merged-local index file, we fallback
// to fetch the original blocks. We do not report block fetch failure
// and will continue with the remaining local block read.
- logWarning(s"Error occurred while reading push-merged-local index, " +
- s"prepare to fetch the original blocks", e)
+ logWarning("Error occurred while reading push-merged-local index, " +
+ "prepare to fetch the original blocks", e)
pushBasedFetchHelper.initiateFallbackFetchForPushMergedBlock(
shuffleBlockId, pushBasedFetchHelper.localShuffleMergerBlockMgrId)
}
@@ -1142,10 +1145,11 @@ final class ShuffleBlockFetcherIterator(
diagnosisResponse
case shuffleBlockChunk: ShuffleBlockChunkId =>
// TODO SPARK-36284 Add shuffle checksum support for push-based shuffle
- val diagnosisResponse = s"BlockChunk $shuffleBlockChunk is corrupted but corruption " +
+ logWarning(log"BlockChunk ${MDC(SHUFFLE_BLOCK_INFO, shuffleBlockChunk)} " +
+ log"is corrupted but corruption diagnosis is skipped due to lack of shuffle " +
+ log"checksum support for push-based shuffle.")
+ s"BlockChunk $shuffleBlockChunk is corrupted but corruption " +
s"diagnosis is skipped due to lack of shuffle checksum support for push-based shuffle."
- logWarning(diagnosisResponse)
- diagnosisResponse
case unexpected: BlockId =>
throw SparkException.internalError(
s"Unexpected type of BlockId, $unexpected", category = "STORAGE")
diff --git a/core/src/main/scala/org/apache/spark/storage/TopologyMapper.scala b/core/src/main/scala/org/apache/spark/storage/TopologyMapper.scala
index 3c2c4b46dc4c..f1dca53c7e3b 100644
--- a/core/src/main/scala/org/apache/spark/storage/TopologyMapper.scala
+++ b/core/src/main/scala/org/apache/spark/storage/TopologyMapper.scala
@@ -19,7 +19,8 @@ package org.apache.spark.storage
import org.apache.spark.SparkConf
import org.apache.spark.annotation.DeveloperApi
-import org.apache.spark.internal.{config, Logging}
+import org.apache.spark.internal.{config, Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.util.Utils
/**
@@ -78,7 +79,7 @@ class FileBasedTopologyMapper(conf: SparkConf) extends TopologyMapper(conf) with
if (topology.isDefined) {
logDebug(s"$hostname -> ${topology.get}")
} else {
- logWarning(s"$hostname does not have any topology information")
+ logWarning(log"${MDC(HOST_PORT, hostname)} does not have any topology information")
}
topology
}
diff --git a/core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala b/core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala
index 48d2ef68b41a..1283b9340a45 100644
--- a/core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala
+++ b/core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala
@@ -30,7 +30,8 @@ import scala.util.control.NonFatal
import com.google.common.io.ByteStreams
import org.apache.spark.{SparkConf, SparkException, TaskContext}
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.{STORAGE_UNROLL_MEMORY_THRESHOLD, UNROLL_MEMORY_CHECK_PERIOD, UNROLL_MEMORY_GROWTH_FACTOR}
import org.apache.spark.memory.{MemoryManager, MemoryMode}
import org.apache.spark.serializer.{SerializationStream, SerializerManager}
@@ -110,9 +111,10 @@ private[spark] class MemoryStore(
}
if (maxMemory < unrollMemoryThreshold) {
- logWarning(s"Max memory ${Utils.bytesToString(maxMemory)} is less than the initial memory " +
- s"threshold ${Utils.bytesToString(unrollMemoryThreshold)} needed to store a block in " +
- s"memory. Please configure Spark with more memory.")
+ logWarning(log"Max memory ${MDC(NUM_BYTES, Utils.bytesToString(maxMemory))} " +
+ log"is less than the initial memory " +
+ log"threshold ${MDC(MAX_SIZE, Utils.bytesToString(unrollMemoryThreshold))} " +
+ log"needed to store a block in memory. Please configure Spark with more memory.")
}
logInfo("MemoryStore started with capacity %s".format(Utils.bytesToString(maxMemory)))
@@ -213,8 +215,9 @@ private[spark] class MemoryStore(
reserveUnrollMemoryForThisTask(blockId, initialMemoryThreshold, memoryMode)
if (!keepUnrolling) {
- logWarning(s"Failed to reserve initial memory threshold of " +
- s"${Utils.bytesToString(initialMemoryThreshold)} for computing block $blockId in memory.")
+ logWarning(log"Failed to reserve initial memory threshold of " +
+ log"${MDC(NUM_BYTES, Utils.bytesToString(initialMemoryThreshold))} " +
+ log"for computing block ${MDC(BLOCK_ID, blockId)} in memory.")
} else {
unrollMemoryUsedByThisBlock += initialMemoryThreshold
}
@@ -348,9 +351,10 @@ private[spark] class MemoryStore(
// Initial per-task memory to request for unrolling blocks (bytes).
val initialMemoryThreshold = unrollMemoryThreshold
val chunkSize = if (initialMemoryThreshold > ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH) {
- logWarning(s"Initial memory threshold of ${Utils.bytesToString(initialMemoryThreshold)} " +
- s"is too large to be set as chunk size. Chunk size has been capped to " +
- s"${Utils.bytesToString(ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH)}")
+ logWarning(log"Initial memory threshold of " +
+ log"${MDC(NUM_BYTES, Utils.bytesToString(initialMemoryThreshold))} " +
+ log"is too large to be set as chunk size. Chunk size has been capped to " +
+ log"${MDC(MAX_SIZE, Utils.bytesToString(ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH))}")
ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH
} else {
initialMemoryThreshold.toInt
@@ -660,8 +664,8 @@ private[spark] class MemoryStore(
*/
private def logUnrollFailureMessage(blockId: BlockId, finalVectorSize: Long): Unit = {
logWarning(
- s"Not enough space to cache $blockId in memory! " +
- s"(computed ${Utils.bytesToString(finalVectorSize)} so far)"
+ log"Not enough space to cache ${MDC(BLOCK_ID, blockId)} in memory! " +
+ log"(computed ${MDC(NUM_BYTES, Utils.bytesToString(finalVectorSize))} so far)"
)
logMemoryUsage()
}
diff --git a/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala b/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
index 5e567a891d58..3d0e477037ea 100644
--- a/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
+++ b/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
@@ -40,7 +40,9 @@ import org.json4s.JValue
import org.json4s.jackson.JsonMethods.{pretty, render}
import org.apache.spark.{SecurityManager, SparkConf, SSLOptions}
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config.UI._
import org.apache.spark.util.Utils
@@ -84,7 +86,8 @@ private[spark] object JettyUtils extends Logging {
case e: IllegalArgumentException =>
response.sendError(HttpServletResponse.SC_BAD_REQUEST, e.getMessage)
case e: Exception =>
- logWarning(s"GET ${request.getRequestURI} failed: $e", e)
+ logWarning(log"GET ${MDC(LogKeys.URI, request.getRequestURI)} failed: " +
+ log"${MDC(ERROR, e)}", e)
throw e
}
}
diff --git a/core/src/main/scala/org/apache/spark/ui/scope/RDDOperationGraph.scala b/core/src/main/scala/org/apache/spark/ui/scope/RDDOperationGraph.scala
index 9a035e0f1e83..ee7f67233bbd 100644
--- a/core/src/main/scala/org/apache/spark/ui/scope/RDDOperationGraph.scala
+++ b/core/src/main/scala/org/apache/spark/ui/scope/RDDOperationGraph.scala
@@ -25,7 +25,8 @@ import scala.xml.Utility
import org.apache.commons.text.StringEscapeUtils
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.rdd.DeterministicLevel
import org.apache.spark.scheduler.StageInfo
import org.apache.spark.storage.StorageLevel
@@ -214,7 +215,8 @@ private[spark] object RDDOperationGraph extends Logging {
case (true, false) => outgoingEdges += e
case (false, true) => incomingEdges += e
// should never happen
- case _ => logWarning(s"Found an orphan edge in stage ${stage.stageId}: $e")
+ case _ => logWarning(log"Found an orphan edge in stage " +
+ log"${MDC(STAGE_ID, stage.stageId)}: ${MDC(ERROR, e)}")
}
}
diff --git a/core/src/main/scala/org/apache/spark/util/AccumulatorV2.scala b/core/src/main/scala/org/apache/spark/util/AccumulatorV2.scala
index c6d8073a0c2f..3237a321f1c3 100644
--- a/core/src/main/scala/org/apache/spark/util/AccumulatorV2.scala
+++ b/core/src/main/scala/org/apache/spark/util/AccumulatorV2.scala
@@ -23,7 +23,8 @@ import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicLong
import org.apache.spark.{InternalAccumulator, SparkContext, TaskContext}
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.scheduler.AccumulableInfo
import org.apache.spark.util.AccumulatorContext.internOption
@@ -276,7 +277,8 @@ private[spark] object AccumulatorContext extends Logging {
// Since we are storing weak references, warn when the underlying data is not valid.
val acc = ref.get
if (acc eq null) {
- logWarning(s"Attempted to access garbage collected accumulator $id")
+ logWarning(log"Attempted to access garbage collected accumulator " +
+ log"${MDC(ACCUMULATOR_ID, id)}")
}
Option(acc)
}
diff --git a/core/src/main/scala/org/apache/spark/util/DependencyUtils.scala b/core/src/main/scala/org/apache/spark/util/DependencyUtils.scala
index 14851d877289..8526a2125458 100644
--- a/core/src/main/scala/org/apache/spark/util/DependencyUtils.scala
+++ b/core/src/main/scala/org/apache/spark/util/DependencyUtils.scala
@@ -26,7 +26,9 @@ import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.deploy.SparkSubmit
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.util.ArrayImplicits._
@@ -223,10 +225,10 @@ private[spark] object DependencyUtils extends Logging {
if (file.exists()) {
loader.addURL(file.toURI.toURL)
} else {
- logWarning(s"Local jar $file does not exist, skipping.")
+ logWarning(log"Local jar ${MDC(FILE_NAME, file)} does not exist, skipping.")
}
case _ =>
- logWarning(s"Skip remote jar $uri.")
+ logWarning(log"Skip remote jar ${MDC(LogKeys.URI, uri)}.")
}
}
diff --git a/core/src/main/scala/org/apache/spark/util/HadoopFSUtils.scala b/core/src/main/scala/org/apache/spark/util/HadoopFSUtils.scala
index a4b9ba7bb016..49c63c905a33 100644
--- a/core/src/main/scala/org/apache/spark/util/HadoopFSUtils.scala
+++ b/core/src/main/scala/org/apache/spark/util/HadoopFSUtils.scala
@@ -28,7 +28,8 @@ import org.apache.hadoop.fs.viewfs.ViewFileSystem
import org.apache.hadoop.hdfs.DistributedFileSystem
import org.apache.spark._
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.metrics.source.HiveCatalogMetrics
import org.apache.spark.util.ArrayImplicits._
@@ -99,7 +100,8 @@ private[spark] object HadoopFSUtils extends Logging {
Seq((path, statues.toImmutableArraySeq))
} catch {
case _: FileNotFoundException =>
- logWarning(s"The root directory $path was not found. Was it deleted very recently?")
+ logWarning(log"The root directory ${MDC(PATH, path)} " +
+ log"was not found. Was it deleted very recently?")
Seq((path, Seq.empty[FileStatus]))
}
}
@@ -235,7 +237,8 @@ private[spark] object HadoopFSUtils extends Logging {
// InMemoryFileIndex construction. However, it's still a net improvement to detect and
// fail-fast on the non-root cases. For more info see the SPARK-27676 review discussion.
case _: FileNotFoundException if isRootPath || ignoreMissingFiles =>
- logWarning(s"The directory $path was not found. Was it deleted very recently?")
+ logWarning(log"The directory ${MDC(PATH, path)} " +
+ log"was not found. Was it deleted very recently?")
Array.empty[FileStatus]
}
@@ -323,8 +326,8 @@ private[spark] object HadoopFSUtils extends Logging {
}
if (missingFiles.nonEmpty) {
- logWarning(
- s"the following files were missing during file scan:\n ${missingFiles.mkString("\n ")}")
+ logWarning(log"the following files were missing during file scan:\n " +
+ log"${MDC(PATHS, missingFiles.mkString("\n "))}")
}
resolvedLeafStatuses.toImmutableArraySeq
diff --git a/core/src/main/scala/org/apache/spark/util/PeriodicCheckpointer.scala b/core/src/main/scala/org/apache/spark/util/PeriodicCheckpointer.scala
index f01645d82303..7a98c4830db9 100644
--- a/core/src/main/scala/org/apache/spark/util/PeriodicCheckpointer.scala
+++ b/core/src/main/scala/org/apache/spark/util/PeriodicCheckpointer.scala
@@ -23,7 +23,8 @@ import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.spark.SparkContext
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
/**
@@ -184,9 +185,9 @@ private[spark] object PeriodicCheckpointer extends Logging {
val fs = path.getFileSystem(conf)
fs.delete(path, true)
} catch {
- case e: Exception =>
- logWarning("PeriodicCheckpointer could not remove old checkpoint file: " +
- checkpointFile)
+ case _: Exception =>
+ logWarning(log"PeriodicCheckpointer could not remove old checkpoint file: " +
+ log"${MDC(FILE_NAME, checkpointFile)}")
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/util/SignalUtils.scala b/core/src/main/scala/org/apache/spark/util/SignalUtils.scala
index 775dc44fc1a1..098f8b698fb7 100644
--- a/core/src/main/scala/org/apache/spark/util/SignalUtils.scala
+++ b/core/src/main/scala/org/apache/spark/util/SignalUtils.scala
@@ -25,7 +25,8 @@ import org.apache.commons.lang3.SystemUtils
import org.slf4j.Logger
import sun.misc.{Signal, SignalHandler}
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC, MessageWithContext}
+import org.apache.spark.internal.LogKeys._
/**
* Contains utilities for working with posix signals.
@@ -58,7 +59,7 @@ private[spark] object SignalUtils extends Logging {
*/
def register(signal: String)(action: => Boolean): Unit = {
if (SystemUtils.IS_OS_UNIX) {
- register(signal, s"Failed to register signal handler for $signal",
+ register(signal, log"Failed to register signal handler for ${MDC(SIGNAL, signal)}",
logStackTrace = true)(action)
}
}
@@ -74,7 +75,7 @@ private[spark] object SignalUtils extends Logging {
*/
def register(
signal: String,
- failMessage: String,
+ failMessage: MessageWithContext,
logStackTrace: Boolean = true)(
action: => Boolean): Unit = synchronized {
try {
diff --git a/core/src/main/scala/org/apache/spark/util/SizeEstimator.scala b/core/src/main/scala/org/apache/spark/util/SizeEstimator.scala
index 1447a3e752de..88fe64859a21 100644
--- a/core/src/main/scala/org/apache/spark/util/SizeEstimator.scala
+++ b/core/src/main/scala/org/apache/spark/util/SizeEstimator.scala
@@ -151,11 +151,11 @@ object SizeEstimator extends Logging {
// TODO: We could use reflection on the VMOption returned ?
getVMMethod.invoke(bean, "UseCompressedOops").toString.contains("true")
} catch {
- case e: Exception =>
+ case _: Exception =>
// Guess whether they've enabled UseCompressedOops based on whether maxMemory < 32 GB
val guess = Runtime.getRuntime.maxMemory < (32L*1024*1024*1024)
- val guessInWords = if (guess) "yes" else "not"
- logWarning("Failed to check whether UseCompressedOops is set; assuming " + guessInWords)
+ logWarning(log"Failed to check whether UseCompressedOops is set; " +
+ log"assuming " + (if (guess) log"yes" else log"not"))
guess
}
}
diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala
index 67ba72485277..8ffe240b95a5 100644
--- a/core/src/main/scala/org/apache/spark/util/Utils.scala
+++ b/core/src/main/scala/org/apache/spark/util/Utils.scala
@@ -69,6 +69,7 @@ import org.slf4j.Logger
import org.apache.spark._
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.internal.{Logging, MDC, MessageWithContext}
+import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.Streaming._
@@ -444,7 +445,8 @@ private[spark] object Utils
// TODO(SPARK-38632): should keep file permissions. Java implementation doesn't.
unTarUsingJava(source, dest)
} else {
- logWarning(s"Cannot unpack $source, just copying it to $dest.")
+ logWarning(log"Cannot unpack ${MDC(LogKeys.FILE_NAME, source)}, " +
+ log"just copying it to ${MDC(FILE_NAME2, dest)}.")
copyRecursive(source, dest)
}
}
@@ -797,8 +799,10 @@ private[spark] object Utils
}
if (uris.nonEmpty) {
logWarning(
- "The configured local directories are not expected to be URIs; however, got suspicious " +
- s"values [${uris.mkString(", ")}]. Please check your configured local directories.")
+ log"The configured local directories are not expected to be URIs; " +
+ log"however, got suspicious values [" +
+ log"${MDC(LogKeys.URIS, uris.mkString(", "))}]. " +
+ log"Please check your configured local directories.")
}
configuredLocalDirs.flatMap { root =>
@@ -889,16 +893,17 @@ private[spark] object Utils
// because of Inet6Address.toHostName may add interface at the end if it knows about it
val strippedAddress = InetAddress.getByAddress(addr.getAddress)
// We've found an address that looks reasonable!
- logWarning("Your hostname, " + InetAddress.getLocalHost.getHostName + " resolves to" +
- " a loopback address: " + address.getHostAddress + "; using " +
- strippedAddress.getHostAddress + " instead (on interface " + ni.getName + ")")
+ logWarning(log"Your hostname, ${MDC(HOST, InetAddress.getLocalHost.getHostName)}, " +
+ log"resolves to a loopback address: ${MDC(HOST_PORT, address.getHostAddress)}; " +
+ log"using ${MDC(HOST_PORT2, strippedAddress.getHostAddress)} instead (on interface " +
+ log"${MDC(NETWORK_IF, ni.getName)})")
logWarning("Set SPARK_LOCAL_IP if you need to bind to another address")
return strippedAddress
}
}
- logWarning("Your hostname, " + InetAddress.getLocalHost.getHostName + " resolves to" +
- " a loopback address: " + address.getHostAddress + ", but we couldn't find any" +
- " external IP address!")
+ logWarning(log"Your hostname, ${MDC(HOST, InetAddress.getLocalHost.getHostName)}, " +
+ log"resolves to a loopback address: ${MDC(HOST_PORT, address.getHostAddress)}, " +
+ log"but we couldn't find any external IP address!")
logWarning("Set SPARK_LOCAL_IP if you need to bind to another address")
}
address
@@ -1330,7 +1335,7 @@ private[spark] object Utils
case t: Throwable =>
if (originalThrowable != t) {
originalThrowable.addSuppressed(t)
- logWarning(s"Suppressing exception in catch: ${t.getMessage}", t)
+ logWarning(log"Suppressing exception in catch: ${MDC(ERROR, t.getMessage)}", t)
}
}
throw originalThrowable
@@ -1340,7 +1345,7 @@ private[spark] object Utils
} catch {
case t: Throwable if (originalThrowable != null && originalThrowable != t) =>
originalThrowable.addSuppressed(t)
- logWarning(s"Suppressing exception in finally: ${t.getMessage}", t)
+ logWarning(log"Suppressing exception in finally: ${MDC(ERROR, t.getMessage)}", t)
throw originalThrowable
}
}
@@ -2185,11 +2190,13 @@ private[spark] object Utils
if (startPort == 0) {
// As startPort 0 is for a random free port, it is most possibly binding address is
// not correct.
- logWarning(s"Service$serviceString could not bind on a random free port. " +
- "You may check whether configuring an appropriate binding address.")
+ logWarning(log"Service${MDC(SERVICE_NAME, serviceString)} " +
+ log"could not bind on a random free port. " +
+ log"You may check whether configuring an appropriate binding address.")
} else {
- logWarning(s"Service$serviceString could not bind on port $tryPort. " +
- s"Attempting port ${tryPort + 1}.")
+ logWarning(log"Service${MDC(SERVICE_NAME, serviceString)} " +
+ log"could not bind on port ${MDC(PORT, tryPort)}. " +
+ log"Attempting port ${MDC(PORT2, tryPort + 1)}.")
}
}
}
@@ -2454,9 +2461,9 @@ private[spark] object Utils
(isShuffleServiceAndYarn || isTesting) && ioEncryptionDisabled && serializerIsSupported
}
if (!canDoPushBasedShuffle) {
- logWarning("Push-based shuffle can only be enabled when the application is submitted " +
- "to run in YARN mode, with external shuffle service enabled, IO encryption disabled, " +
- "and relocation of serialized objects supported.")
+ logWarning(log"Push-based shuffle can only be enabled when the application is submitted " +
+ log"to run in YARN mode, with external shuffle service enabled, IO encryption " +
+ log"disabled, and relocation of serialized objects supported.")
}
canDoPushBasedShuffle
@@ -2517,15 +2524,15 @@ private[spark] object Utils
*/
def getDynamicAllocationInitialExecutors(conf: SparkConf): Int = {
if (conf.get(DYN_ALLOCATION_INITIAL_EXECUTORS) < conf.get(DYN_ALLOCATION_MIN_EXECUTORS)) {
- logWarning(s"${DYN_ALLOCATION_INITIAL_EXECUTORS.key} less than " +
- s"${DYN_ALLOCATION_MIN_EXECUTORS.key} is invalid, ignoring its setting, " +
- "please update your configs.")
+ logWarning(log"${MDC(CONFIG, DYN_ALLOCATION_INITIAL_EXECUTORS.key)} less than " +
+ log"${MDC(CONFIG2, DYN_ALLOCATION_MIN_EXECUTORS.key)} is invalid, ignoring its setting, " +
+ log"please update your configs.")
}
if (conf.get(EXECUTOR_INSTANCES).getOrElse(0) < conf.get(DYN_ALLOCATION_MIN_EXECUTORS)) {
- logWarning(s"${EXECUTOR_INSTANCES.key} less than " +
- s"${DYN_ALLOCATION_MIN_EXECUTORS.key} is invalid, ignoring its setting, " +
- "please update your configs.")
+ logWarning(log"${MDC(CONFIG, EXECUTOR_INSTANCES.key)} less than " +
+ log"${MDC(CONFIG2, DYN_ALLOCATION_MIN_EXECUTORS.key)} is invalid, ignoring its setting, " +
+ log"please update your configs.")
}
val initialExecutors = Seq(
@@ -2758,7 +2765,7 @@ private[spark] object Utils
case Some("https") =>
masterWithoutK8sPrefix
case Some("http") =>
- logWarning("Kubernetes master URL uses HTTP instead of HTTPS.")
+ logWarning(log"Kubernetes master URL uses HTTP instead of HTTPS.")
masterWithoutK8sPrefix
case _ =>
throw new IllegalArgumentException("Invalid Kubernetes master scheme: " + masterScheme
@@ -3118,7 +3125,8 @@ private[spark] class CallerContext(
context
} else {
val finalContext = context.substring(0, len)
- logWarning(s"Truncated Spark caller context from $context to $finalContext")
+ logWarning(log"Truncated Spark caller context from ${MDC(CONTEXT, context)} " +
+ log"to ${MDC(FINAL_CONTEXT, finalContext)}")
finalContext
}
}
diff --git a/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala b/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
index 8224472b7545..572e69885325 100644
--- a/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
+++ b/core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala
@@ -29,7 +29,8 @@ import com.google.common.io.ByteStreams
import org.apache.spark.{SparkEnv, TaskContext}
import org.apache.spark.annotation.DeveloperApi
import org.apache.spark.executor.ShuffleWriteMetrics
-import org.apache.spark.internal.{config, Logging}
+import org.apache.spark.internal.{config, Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.serializer.{DeserializationStream, Serializer, SerializerManager}
import org.apache.spark.storage.{BlockId, BlockManager}
import org.apache.spark.util.CompletionIterator
@@ -544,7 +545,7 @@ class ExternalAppendOnlyMap[K, V, C](
}
if (file.exists()) {
if (!file.delete()) {
- logWarning(s"Error deleting ${file}")
+ logWarning(log"Error deleting ${MDC(FILE_NAME, file)}")
}
}
}
diff --git a/core/src/main/scala/org/apache/spark/util/logging/DriverLogger.scala b/core/src/main/scala/org/apache/spark/util/logging/DriverLogger.scala
index ffb13db51582..2682655f259f 100644
--- a/core/src/main/scala/org/apache/spark/util/logging/DriverLogger.scala
+++ b/core/src/main/scala/org/apache/spark/util/logging/DriverLogger.scala
@@ -33,7 +33,8 @@ import org.apache.logging.log4j.core.layout.PatternLayout
import org.apache.spark.SparkConf
import org.apache.spark.deploy.SparkHadoopUtil
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.network.util.JavaUtils
import org.apache.spark.util.{ThreadUtils, Utils}
@@ -229,8 +230,8 @@ private[spark] object DriverLogger extends Logging {
// Return None because we don't need DFS-related logic in SparkContext and DfsAsyncWriter
None
} else {
- logWarning(s"Driver logs are not persisted because" +
- s" ${DRIVER_LOG_DFS_DIR.key} is not configured")
+ logWarning(log"Driver logs are not persisted because" +
+ log" ${MDC(CONFIG, DRIVER_LOG_DFS_DIR.key)} is not configured")
None
}
} else {
diff --git a/core/src/main/scala/org/apache/spark/util/logging/FileAppender.scala b/core/src/main/scala/org/apache/spark/util/logging/FileAppender.scala
index 23cd222f68ef..f33af946bd6e 100644
--- a/core/src/main/scala/org/apache/spark/util/logging/FileAppender.scala
+++ b/core/src/main/scala/org/apache/spark/util/logging/FileAppender.scala
@@ -21,7 +21,7 @@ import java.io.{File, FileOutputStream, InputStream, IOException}
import org.apache.spark.SparkConf
import org.apache.spark.internal.{config, Logging, MDC}
-import org.apache.spark.internal.LogKeys.PATH
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.util.{IntParam, Utils}
/**
@@ -150,8 +150,9 @@ private[spark] object FileAppender extends Logging {
logInfo(s"Rolling executor logs enabled for $file with rolling $seconds seconds")
Some((seconds * 1000L, "--yyyy-MM-dd--HH-mm-ss"))
case _ =>
- logWarning(s"Illegal interval for rolling executor logs [$rollingInterval], " +
- s"rolling logs not enabled")
+ logWarning(log"Illegal interval for rolling executor logs [" +
+ log"${MDC(TIME_UNITS, rollingInterval)}], " +
+ log"rolling logs not enabled")
None
}
validatedParams.map {
@@ -172,7 +173,8 @@ private[spark] object FileAppender extends Logging {
inputStream, file, new SizeBasedRollingPolicy(bytes), conf, closeStreams = closeStreams)
case _ =>
logWarning(
- s"Illegal size [$rollingSizeBytes] for rolling executor logs, rolling logs not enabled")
+ log"Illegal size [${MDC(NUM_BYTES, rollingSizeBytes)}] " +
+ log"for rolling executor logs, rolling logs not enabled")
new FileAppender(inputStream, file, closeStreams = closeStreams)
}
}
diff --git a/core/src/main/scala/org/apache/spark/util/logging/RollingFileAppender.scala b/core/src/main/scala/org/apache/spark/util/logging/RollingFileAppender.scala
index 28ffdc850897..465f07297654 100644
--- a/core/src/main/scala/org/apache/spark/util/logging/RollingFileAppender.scala
+++ b/core/src/main/scala/org/apache/spark/util/logging/RollingFileAppender.scala
@@ -25,7 +25,7 @@ import org.apache.commons.io.IOUtils
import org.apache.spark.SparkConf
import org.apache.spark.internal.{config, MDC}
-import org.apache.spark.internal.LogKeys.PATH
+import org.apache.spark.internal.LogKeys._
import org.apache.spark.util.ArrayImplicits._
/**
@@ -132,12 +132,13 @@ private[spark] class RollingFileAppender(
i += 1
} while (i < 10000 && rolloverFileExist(altRolloverFile))
- logWarning(s"Rollover file $rolloverFile already exists, " +
- s"rolled over $activeFile to file $altRolloverFile")
+ logWarning(log"Rollover file ${MDC(FILE_NAME, rolloverFile)} already exists, " +
+ log"rolled over ${MDC(FILE_NAME2, activeFile)} " +
+ log"to file ${MDC(FILE_NAME3, altRolloverFile)}")
rotateFile(activeFile, altRolloverFile)
}
} else {
- logWarning(s"File $activeFile does not exist")
+ logWarning(log"File ${MDC(FILE_NAME, activeFile)} does not exist")
}
}
diff --git a/core/src/main/scala/org/apache/spark/util/logging/RollingPolicy.scala b/core/src/main/scala/org/apache/spark/util/logging/RollingPolicy.scala
index 5327ecd3e56a..310e89593094 100644
--- a/core/src/main/scala/org/apache/spark/util/logging/RollingPolicy.scala
+++ b/core/src/main/scala/org/apache/spark/util/logging/RollingPolicy.scala
@@ -20,7 +20,8 @@ package org.apache.spark.util.logging
import java.text.SimpleDateFormat
import java.util.{Calendar, Locale}
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MDC}
+import org.apache.spark.internal.LogKeys._
/**
* Defines the policy based on which [[org.apache.spark.util.logging.RollingFileAppender]] will
@@ -53,8 +54,9 @@ private[spark] class TimeBasedRollingPolicy(
import TimeBasedRollingPolicy._
if (checkIntervalConstraint && rolloverIntervalMillis < MINIMUM_INTERVAL_SECONDS * 1000L) {
- logWarning(s"Rolling interval [${rolloverIntervalMillis/1000L} seconds] is too small. " +
- s"Setting the interval to the acceptable minimum of $MINIMUM_INTERVAL_SECONDS seconds.")
+ logWarning(log"Rolling interval [${MDC(TIME_UNITS, rolloverIntervalMillis)} " +
+ log"ms] is too small. Setting the interval to the acceptable minimum of " +
+ log"${MDC(MIN_TIME, MINIMUM_INTERVAL_SECONDS * 1000)} ms.")
rolloverIntervalMillis = MINIMUM_INTERVAL_SECONDS * 1000L
}
@@ -103,8 +105,9 @@ private[spark] class SizeBasedRollingPolicy(
import SizeBasedRollingPolicy._
if (checkSizeConstraint && rolloverSizeBytes < MINIMUM_SIZE_BYTES) {
- logWarning(s"Rolling size [$rolloverSizeBytes bytes] is too small. " +
- s"Setting the size to the acceptable minimum of $MINIMUM_SIZE_BYTES bytes.")
+ logWarning(log"Rolling size [${MDC(NUM_BYTES, rolloverSizeBytes)} bytes] is too small. " +
+ log"Setting the size to the acceptable minimum of ${MDC(MIN_SIZE, MINIMUM_SIZE_BYTES)} " +
+ log"bytes.")
rolloverSizeBytes = MINIMUM_SIZE_BYTES
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org