dont use case classes for status api POJOs, since they have binary compatibility issues

Author: squito

core/src/main/scala/org/apache/spark/status/api/v1/AllJobsResource.scala

@@ -72,7 +72,7 @@ object AllJobsResource {
       }
       val lastStageName = lastStageInfo.map { _.name }.getOrElse("(Unknown Stage Name)")
       val lastStageDescription = lastStageData.flatMap { _.description }
-      JobData(
+      new JobData(
         jobId = job.jobId,
         name = lastStageName,
         description = lastStageDescription,

core/src/main/scala/org/apache/spark/status/api/v1/AllRDDResource.scala

@@ -73,7 +73,7 @@ object AllRDDResource {
 
     val dataDistribution = if (includeDetails) {
       Some(storageStatusList.map { status =>
-        RDDDataDistribution(
+        new RDDDataDistribution(
           address = status.blockManagerId.hostPort,
           memoryUsed = status.memUsedByRdd(rddId),
           memoryRemaining = status.memRemaining,
@@ -84,7 +84,7 @@ object AllRDDResource {
     }
     val partitions = if (includeDetails) {
       Some(blocks.map { case(id, block, locations) =>
-        RDDPartitionInfo(
+        new RDDPartitionInfo(
           blockName = id.name,
           storageLevel = block.storageLevel.description,
           memoryUsed = block.memSize,
@@ -96,7 +96,7 @@ object AllRDDResource {
       None
     }
 
-    RDDStorageInfo(
+    new RDDStorageInfo(
       id = rddId,
       name = rddInfo.name,
       numPartitions = rddInfo.numPartitions,

core/src/main/scala/org/apache/spark/status/api/v1/AllStagesResource.scala

@@ -72,7 +72,7 @@ object AllStagesResource {
     }
     val executorSummary = if(includeDetails) {
       Some(stageUiData.executorSummary.map { case(k,summary) =>
-        k -> ExecutorStageSummary(
+        k -> new ExecutorStageSummary(
           taskTime = summary.taskTime,
           failedTasks = summary.failedTasks,
           succeededTasks = summary.succeededTasks,
@@ -87,7 +87,7 @@ object AllStagesResource {
     } else {
       None
     }
-    StageData(
+    new StageData(
       status = status,
       stageId = stageInfo.stageId,
       numActiveTasks = stageUiData.numActiveTasks,
@@ -126,7 +126,7 @@ object AllStagesResource {
 
 
   def convertTaskData(uiData: TaskUIData): TaskData = {
-    TaskData(
+    new TaskData(
       taskId = uiData.taskInfo.taskId,
       index = uiData.taskInfo.index,
       attempt = uiData.taskInfo.attempt,
@@ -141,7 +141,7 @@ object AllStagesResource {
   }
 
   def convertUiTaskMetrics(internal: InternalTaskMetrics): TaskMetrics = {
-    TaskMetrics(
+    new TaskMetrics(
       executorDeserializeTime = internal.executorDeserializeTime,
       executorRunTime = internal.executorRunTime,
       resultSize = internal.resultSize,
@@ -157,21 +157,21 @@ object AllStagesResource {
   }
 
   def convertInputMetrics(internal: InternalInputMetrics): InputMetrics = {
-    InputMetrics(
+    new InputMetrics(
       bytesRead = internal.bytesRead,
       recordsRead = internal.recordsRead
     )
   }
 
   def convertOutputMetrics(internal: InternalOutputMetrics): OutputMetrics = {
-    OutputMetrics(
+    new OutputMetrics(
       bytesWritten = internal.bytesWritten,
       recordsWritten = internal.recordsWritten
     )
   }
 
   def convertShuffleReadMetrics(internal: InternalShuffleReadMetrics): ShuffleReadMetrics = {
-    ShuffleReadMetrics(
+    new ShuffleReadMetrics(
       remoteBlocksFetched = internal.remoteBlocksFetched,
       localBlocksFetched = internal.localBlocksFetched,
       fetchWaitTime = internal.fetchWaitTime,
@@ -182,7 +182,7 @@ object AllStagesResource {
   }
 
   def convertShuffleWriteMetrics(internal: InternalShuffleWriteMetrics): ShuffleWriteMetrics = {
-    ShuffleWriteMetrics(
+    new ShuffleWriteMetrics(
       bytesWritten = internal.shuffleBytesWritten,
       writeTime = internal.shuffleWriteTime,
       recordsWritten = internal.shuffleRecordsWritten

core/src/main/scala/org/apache/spark/status/api/v1/ApplicationListResource.scala

@@ -55,7 +55,7 @@ class ApplicationListResource(uiRoot: UIRoot) {
 
 object ApplicationsListResource {
   def appHistoryInfoToPublicAppInfo(app: ApplicationHistoryInfo): ApplicationInfo = {
-    ApplicationInfo(
+    new ApplicationInfo(
       id = app.id,
       name = app.name,
       startTime = new Date(app.startTime),
@@ -68,7 +68,7 @@ object ApplicationsListResource {
   def convertApplicationInfo(
       internal: InternalApplicationInfo,
       completed: Boolean): ApplicationInfo = {
-    ApplicationInfo(
+    new ApplicationInfo(
       id = internal.id,
       name = internal.desc.name,
       startTime = new Date(internal.startTime),