Update pom to fail on warnings (NVIDIA#701)

Update pom to fail on warnings

Signed-off-by: Ahmed Hussein (amahussein) <[email protected]>

core/pom.xml

@@ -446,6 +446,8 @@
  <maven-compiler-plugin.version>3.11.0</maven-compiler-plugin.version>
  <maven.compiler.source>${java.version}</maven.compiler.source>
  <maven.compiler.target>${java.version}</maven.compiler.target>
+ <maven.scaladoc.skip>false</maven.scaladoc.skip>
+ <maven.scalastyle.skip>false</maven.scalastyle.skip>
  <project.build.sourceEncoding>${platform-encoding}</project.build.sourceEncoding>
  <project.reporting.sourceEncoding>${platform-encoding}</project.reporting.sourceEncoding>
  <project.reporting.outputEncoding>${platform-encoding}</project.reporting.outputEncoding>
@@ -702,8 +704,10 @@
  <arg>-feature</arg>
  <arg>-explaintypes</arg>
  <arg>-Yno-adapted-args</arg>
- <arg>-Ywarn-unused:imports</arg>
+ <arg>-Ywarn-unused:imports,locals,patvars,privates</arg>
  <arg>-Xlint:missing-interpolator</arg>
+ <arg>-Xfatal-warnings</arg>
+ <arg>-Wconf:cat=lint-adapted-args:e</arg>
  </args>
  <jvmArgs>
  <jvmArg>-Xms1024m</jvmArg>

core/src/main/scala/com/nvidia/spark/rapids/tool/EventLogPathProcessor.scala

@@ -185,7 +185,7 @@ object EventLogPathProcessor extends Logging {
  }.toMap
  }
  } catch {
- case fe: FileNotFoundException =>
+ case _: FileNotFoundException =>
  logWarning(s"$pathString not found, skipping!")
  Map.empty[EventLogInfo, Long]
  case e: Exception =>

core/src/main/scala/com/nvidia/spark/rapids/tool/Platform.scala

@@ -133,8 +133,8 @@ object PlatformFactory extends Logging {
  *
  * @param platformKey The key representing the desired platform.
  * @return An instance of the specified platform.
- * @throws IllegalArgumentException if the specified platform key is not supported.
  */
+ @throws[IllegalArgumentException]
  @tailrec
  def createInstance(platformKey: String = PlatformNames.DEFAULT): Platform = {
  platformKey match {

core/src/main/scala/com/nvidia/spark/rapids/tool/planparser/DataWritingCommandExecParser.scala

@@ -57,13 +57,11 @@ object DataWritingCommandExecParser {
  val saveIntoDataSrcCMD = "SaveIntoDataSourceCommand"
  val insertIntoHadoopCMD = "InsertIntoHadoopFsRelationCommand"
 
- // List of writeExecs that represent a physical command.
+ // Note: List of writeExecs that represent a physical command.
  // hardcode because InsertIntoHadoopFsRelationCommand uses this same exec
  // and InsertIntoHadoopFsRelationCommand doesn't have an entry in the
- // supported execs file
- private val physicalWriteCommands = Set(
- defaultPhysicalCMD
- )
+ // supported execs file Set(defaultPhysicalCMD)
+
 
  // A set of the logical commands that will be mapped to the physical write command
  // which has an entry in the speedupSheet
@@ -76,13 +74,9 @@ object DataWritingCommandExecParser {
  saveIntoDataSrcCMD
  )
 
- // Defines a list of the execs that include formatted data.
+ // Note: Defines a list of the execs that include formatted data.
  // This will be used to extract the format and then check whether the
- // format is supported or not.
- private val formattedWriteCommands = Set(
- dataWriteCMD,
- insertIntoHadoopCMD
- )
+ // format is supported or not. Set(dataWriteCMD, insertIntoHadoopCMD)
 
  // For now, we map the SaveIntoDataSourceCommand to defaultPhysicalCMD because we do not
  // have speedup entry for the deltaLake write operation

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/Analysis.scala

@@ -99,15 +99,15 @@ class Analysis(apps: Seq[ApplicationInfo]) {
  }
  }
  val allJobStageRows = apps.flatMap { app =>
- app.jobIdToInfo.flatMap { case (id, jc) =>
+ app.jobIdToInfo.flatMap { case (_, jc) =>
  val stageIdsInJob = jc.stageIds
  val stagesInJob = app.stageIdToInfo.filterKeys { case (sid, _) =>
  stageIdsInJob.contains(sid)
  }
  if (stagesInJob.isEmpty) {
  None
  } else {
- stagesInJob.map { case ((id, said), sc) =>
+ stagesInJob.map { case ((id, _), sc) =>
  val tasksInStage = app.taskEnd.filter { tc =>
  tc.stageId == id
  }
@@ -153,7 +153,7 @@ class Analysis(apps: Seq[ApplicationInfo]) {
  }
  // stages that are missing from a job, perhaps dropped events
  val stagesWithoutJobs = apps.flatMap { app =>
- val allStageinJobs = app.jobIdToInfo.flatMap { case (id, jc) =>
+ val allStageinJobs = app.jobIdToInfo.flatMap { case (_, jc) =>
  val stageIdsInJob = jc.stageIds
  app.stageIdToInfo.filterKeys { case (sid, _) =>
  stageIdsInJob.contains(sid)
@@ -224,15 +224,13 @@ class Analysis(apps: Seq[ApplicationInfo]) {
  } else {
  Seq.empty
  }
-
  }
 
  // SQL Level TaskMetrics Aggregation(Only when SQL exists)
  def sqlMetricsAggregation(): Seq[SQLTaskAggMetricsProfileResult] = {
  val allRows = apps.flatMap { app =>
  app.sqlIdToInfo.map { case (sqlId, sqlCase) =>
  val jcs = app.jobIdToInfo.filter { case (_, jc) =>
- val jcid = jc.sqlID.getOrElse(-1)
  jc.sqlID.getOrElse(-1) == sqlId
  }
  if (jcs.isEmpty) {
@@ -431,10 +429,6 @@ class Analysis(apps: Seq[ApplicationInfo]) {
  }
  }
 
- val groupedTasks = tasksWithSkew.groupBy { tc =>
- (tc.stageId, tc.stageAttemptId)
- }
-
  tasksWithSkew.map { tc =>
  val avgShuffleDur = avgsStageInfos.get((tc.stageId, tc.stageAttemptId))
  avgShuffleDur match {

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/CollectInformation.scala

@@ -147,8 +147,6 @@ class CollectInformation(apps: Seq[ApplicationInfo]) extends Logging {
  val rp = app.resourceProfIdToInfo.get(rpId)
  val execMem = rp.map(_.executorResources.get(ResourceProfile.MEMORY)
  .map(_.amount).getOrElse(0L))
- val execCores = rp.map(_.executorResources.get(ResourceProfile.CORES)
- .map(_.amount).getOrElse(0L))
  val execGpus = rp.map(_.executorResources.get("gpu")
  .map(_.amount).getOrElse(0L))
  val taskCpus = rp.map(_.taskResources.get(ResourceProfile.CPUS)
@@ -177,7 +175,7 @@ class CollectInformation(apps: Seq[ApplicationInfo]) extends Logging {
  // get job related information
  def getJobInfo: Seq[JobInfoProfileResult] = {
  val allRows = apps.flatMap { app =>
- app.jobIdToInfo.map { case (jobId, j) =>
+ app.jobIdToInfo.map { case (_, j) =>
  JobInfoProfileResult(app.index, j.jobID, j.stageIds, j.sqlID, j.startTime, j.endTime)
  }
  }
@@ -270,7 +268,6 @@ object CollectInformation extends Logging {
  app.allSQLMetrics.map { metric =>
  val sqlId = metric.sqlID
  val jobsForSql = app.jobIdToInfo.filter { case (_, jc) =>
- val jcid = jc.sqlID.getOrElse(-1)
  jc.sqlID.getOrElse(-1) == sqlId
  }
  val stageIdsForSQL = jobsForSql.flatMap(_._2.stageIds).toSeq

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/Profiler.scala

@@ -102,7 +102,7 @@ class Profiler(hadoopConf: Configuration, appArgs: ProfileArgs, enablePB: Boolea
  // combine them into single tables in the output.
  val profileOutputWriter = new ProfileOutputWriter(s"$outputDir/combined",
  Profiler.COMBINED_LOG_FILE_NAME_PREFIX, numOutputRows, outputCSV = outputCSV)
- val sums = createAppsAndSummarize(eventLogInfos, false, profileOutputWriter)
+ val sums = createAppsAndSummarize(eventLogInfos, profileOutputWriter)
  writeSafelyToOutput(profileOutputWriter, sums, outputCombined)
  profileOutputWriter.close()
  }
@@ -159,7 +159,6 @@ class Profiler(hadoopConf: Configuration, appArgs: ProfileArgs, enablePB: Boolea
  }
 
  private def createApps(allPaths: Seq[EventLogInfo]): Seq[ApplicationInfo] = {
- var errorCodes = ArrayBuffer[Int]()
  val allApps = new ConcurrentLinkedQueue[ApplicationInfo]()
 
  class ProfileThread(path: EventLogInfo, index: Int) extends Runnable {
@@ -201,9 +200,7 @@ class Profiler(hadoopConf: Configuration, appArgs: ProfileArgs, enablePB: Boolea
  }
 
  private def createAppsAndSummarize(allPaths: Seq[EventLogInfo],
- printPlans: Boolean,
  profileOutputWriter: ProfileOutputWriter): Seq[ApplicationSummaryInfo] = {
- var errorCodes = ArrayBuffer[Int]()
  val allApps = new ConcurrentLinkedQueue[ApplicationSummaryInfo]()
 
  class ProfileThread(path: EventLogInfo, index: Int) extends Runnable {
@@ -253,7 +250,7 @@ class Profiler(hadoopConf: Configuration, appArgs: ProfileArgs, enablePB: Boolea
 
  private def createAppAndProcess(
  allPaths: Seq[EventLogInfo],
- startIndex: Int = 1): Unit = {
+ startIndex: Int): Unit = {
  class ProfileProcessThread(path: EventLogInfo, index: Int) extends Runnable {
  def run: Unit = {
  try {
@@ -309,7 +306,7 @@ class Profiler(hadoopConf: Configuration, appArgs: ProfileArgs, enablePB: Boolea
  logInfo(s"Took ${endTime - startTime}ms to process ${path.eventLog.toString}")
  Some(app)
  } catch {
- case json: com.fasterxml.jackson.core.JsonParseException =>
+ case _: com.fasterxml.jackson.core.JsonParseException =>
  logWarning(s"Error parsing JSON: $path")
  None
  case il: IllegalArgumentException =>

core/src/main/scala/com/nvidia/spark/rapids/tool/qualification/PluginTypeChecker.scala

@@ -37,9 +37,6 @@ class PluginTypeChecker(platform: Platform = PlatformFactory.createInstance(),
  speedupFactorFile: Option[String] = None) extends Logging {
 
  private val NS = "NS"
- private val PS = "PS"
- private val PSPART = "PS*"
- private val SPART = "S*"
  // configured off
  private val CO = "CO"
  private val NA = "NA"
@@ -231,7 +228,7 @@ class PluginTypeChecker(platform: Platform = PlatformFactory.createInstance(),
  case "float" => Seq("real")
  case "decimal" => Seq("dec", "numeric")
  case "calendar" => Seq("interval")
- case other => Seq.empty[String]
+ case _ => Seq.empty[String]
  }
  }
 

core/src/main/scala/com/nvidia/spark/rapids/tool/qualification/QualOutputWriter.scala

@@ -520,8 +520,8 @@ object QualOutputWriter {
 
  private def constructOutputRow(
  strAndSizes: Buffer[(String, Int)],
- delimiter: String = TEXT_DELIMITER,
- prettyPrint: Boolean = false): String = {
+ delimiter: String,
+ prettyPrint: Boolean): String = {
  val entireHeader = new StringBuffer
  if (prettyPrint) {
  entireHeader.append(delimiter)
@@ -866,7 +866,7 @@ object QualOutputWriter {
  private def constructExecInfoBuffer(
  info: ExecInfo,
  appId: String,
- delimiter: String = TEXT_DELIMITER,
+ delimiter: String,
  prettyPrint: Boolean,
  headersAndSizes: LinkedHashMap[String, Int],
  reformatCSV: Boolean = true): String = {
@@ -1125,7 +1125,7 @@ object QualOutputWriter {
  private def constructDetailedAppInfoCSVRow(
  appInfo: FormattedQualificationSummaryInfo,
  headersAndSizes: LinkedHashMap[String, Int],
- reportReadSchema: Boolean = false,
+ reportReadSchema: Boolean,
  reformatCSV: Boolean = true): ListBuffer[(String, Int)] = {
  val reformatCSVFunc : String => String =
  if (reformatCSV) str => StringUtils.reformatCSVString(str) else str => stringIfempty(str)
@@ -1198,7 +1198,7 @@ object QualOutputWriter {
  private def constructStatusReportInfo(
  statusInfo: StatusSummaryInfo,
  headersAndSizes: LinkedHashMap[String, Int],
- delimiter: String = TEXT_DELIMITER,
+ delimiter: String,
  prettyPrint: Boolean,
  reformatCSV: Boolean = true): Seq[String] = {
  val reformatCSVFunc: String => String =

core/src/main/scala/com/nvidia/spark/rapids/tool/qualification/RunningQualificationApp.scala

@@ -155,7 +155,8 @@ class RunningQualificationApp(
  val sqlInfo = aggregatePerSQLStats(sqlID)
  val csvResult =
  constructPerSqlResult(sqlInfo, QualOutputWriter.CSV_DELIMITER, false, escapeCSV = true)
- val textResult = constructPerSqlResult(sqlInfo, QualOutputWriter.TEXT_DELIMITER, true)
+ val textResult = constructPerSqlResult(sqlInfo, QualOutputWriter.TEXT_DELIMITER,
+ prettyPrint = true)
  (csvResult, textResult)
  }
 
@@ -176,8 +177,8 @@ class RunningQualificationApp(
 
  private def constructPerSqlResult(
  sqlInfo: Option[EstimatedPerSQLSummaryInfo],
- delimiter: String = "|",
- prettyPrint: Boolean = true,
+ delimiter: String,
+ prettyPrint: Boolean,
  sqlDescLength: Int = SQL_DESC_LENGTH,
  escapeCSV: Boolean = false): String = {
  sqlInfo match {

core/src/main/scala/org/apache/spark/sql/rapids/tool/AppBase.scala

@@ -203,9 +203,9 @@ abstract class AppBase(
  val fs = eventLogPath.getFileSystem(hconf)
  var totalNumEvents = 0
  val readerOpt = eventLog match {
- case dblog: DatabricksEventLog =>
+ case _: DatabricksEventLog =>
  Some(new DatabricksRollingEventLogFilesFileReader(fs, eventLogPath))
- case apachelog => EventLogFileReader(fs, eventLogPath)
+ case _ => EventLogFileReader(fs, eventLogPath)
  }
 
  if (readerOpt.isDefined) {

core/src/main/scala/org/apache/spark/sql/rapids/tool/AppFilterImpl.scala

@@ -311,7 +311,7 @@ object AppFilterImpl {
  val timeInt = try {
  timeStr.toInt
  } catch {
- case ne: NumberFormatException =>
+ case _: NumberFormatException =>
  throw new IllegalArgumentException(s"Invalid time period $appStartStr specified, " +
  "time must be greater than 0 and valid periods are min(minute),h(hours)" +
  ",d(days),w(weeks),m(months).")

core/src/main/scala/org/apache/spark/sql/rapids/tool/profiling/ApplicationInfo.scala

@@ -330,7 +330,7 @@ class ApplicationInfo(
  }
 
  def aggregateSQLStageInfo: Seq[SQLStageInfoProfileResult] = {
- val jobsWithSQL = jobIdToInfo.filter { case (id, j) =>
+ val jobsWithSQL = jobIdToInfo.filter { case (_, j) =>
  j.sqlID.nonEmpty
  }
  val sqlToStages = jobsWithSQL.flatMap { case (jobId, j) =>
@@ -360,7 +360,7 @@ class ApplicationInfo(
  val res = this.appInfo
 
  val estimatedResult = this.appEndTime match {
- case Some(t) => this.appEndTime
+ case Some(_) => this.appEndTime
  case None =>
  val jobEndTimes = jobIdToInfo.map { case (_, jc) => jc.endTime }.filter(_.isDefined)
  val sqlEndTimes = sqlIdToInfo.map { case (_, sc) => sc.endTime }.filter(_.isDefined)

core/src/main/scala/org/apache/spark/sql/rapids/tool/qualification/QualificationAppInfo.scala

@@ -16,8 +16,6 @@
 
 package org.apache.spark.sql.rapids.tool.qualification
 
-import java.util.concurrent.TimeUnit
-
 import scala.collection.mutable.{ArrayBuffer, HashMap}
 
 import com.nvidia.spark.rapids.tool.EventLogInfo
@@ -31,7 +29,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
 import org.apache.spark.sql.execution.SparkPlanInfo
 import org.apache.spark.sql.execution.ui.SparkPlanGraph
-import org.apache.spark.sql.rapids.tool.{AppBase, GpuEventLogException, IgnoreExecs, SupportedMLFuncsName, ToolUtils}
+import org.apache.spark.sql.rapids.tool.{AppBase, GpuEventLogException, SupportedMLFuncsName, ToolUtils}
 
 class QualificationAppInfo(
  eventLogInfo: Option[EventLogInfo],
@@ -106,7 +104,7 @@ class QualificationAppInfo(
  if (startTime > 0) {
  val estimatedResult =
  this.appEndTime match {
- case Some(t) => this.appEndTime
+ case Some(_) => this.appEndTime
  case None =>
  if (lastSQLEndTime.isEmpty && lastJobEndTime.isEmpty) {
  None
@@ -312,7 +310,7 @@ class QualificationAppInfo(
  }
  val transitionsTime = numTransitions match {
  case 0 => 0L // no transitions
- case gpuCpuTransitions =>
+ case _ =>
  // Duration to transfer data from GPU to CPU and vice versa.
  // Assuming it's a PCI-E Gen3, but also assuming that some of the result could be
  // spilled to disk.
@@ -323,13 +321,11 @@ class QualificationAppInfo(
  }
  if (totalBytesRead > 0) {
  val transitionTime = (totalBytesRead /
- QualificationAppInfo.CPU_GPU_TRANSFER_RATE.toDouble) * gpuCpuTransitions
+ QualificationAppInfo.CPU_GPU_TRANSFER_RATE.toDouble) * numTransitions
  (transitionTime * 1000).toLong // convert to milliseconds
  } else {
  0L
  }
-
- case _ => 0L
  }
  val finalEachStageUnsupported = if (transitionsTime != 0) {
  // Add 50% penalty for unsupported duration if there are transitions. This number

core/src/main/scala/org/apache/spark/sql/rapids/tool/qualification/RunningQualificationEventProcessor.scala

@@ -241,7 +241,7 @@ class RunningQualificationEventProcessor(sparkConf: SparkConf) extends SparkList
  listener.onOtherEvent(event)
  event match {
  case e: SparkListenerSQLExecutionStart =>
- logDebug("Starting new SQL query")
+ logDebug(s"Starting new SQL query: ${e.executionId}")
  case e: SparkListenerSQLExecutionEnd =>
  writeSQLDetails(e.executionId)
  case _ =>

core/src/main/scala/org/apache/spark/sql/rapids/tool/util/EventUtils.scala

@@ -34,8 +34,8 @@ object EventUtils extends Logging {
  *
  * @param data value stored in the (value/update) of the AccumulableInfo
  * @return valid parsed long of the content or the duration
- * @throws java.lang.NullPointerException if the argument is `null`
  */
+ @throws[NullPointerException]
  def parseAccumFieldToLong(data: Any): Option[Long] = {
  val strData = data.toString
  try {

core/src/test/scala/com/nvidia/spark/rapids/tool/ToolTestUtils.scala

@@ -137,7 +137,7 @@ object ToolTestUtils extends Logging {
 
  def processProfileApps(logs: Array[String],
  sparkSession: SparkSession): ArrayBuffer[ApplicationInfo] = {
- var apps: ArrayBuffer[ApplicationInfo] = ArrayBuffer[ApplicationInfo]()
+ val apps: ArrayBuffer[ApplicationInfo] = ArrayBuffer[ApplicationInfo]()
  val appArgs = new ProfileArgs(logs)
  var index: Int = 1
  for (path <- appArgs.eventlog()) {