Skip to content
Snippets Groups Projects
Commit fb572c6e authored by Josh Rosen's avatar Josh Rosen
Browse files

[SPARK-12525] Fix fatal compiler warnings in Kinesis ASL due to @transient annotations

The Scala 2.11 SBT build currently fails for Spark 1.6.0 and master due to warnings about the `transient` annotation:

```
[error] [warn] /Users/joshrosen/Documents/spark/extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisBackedBlockRDD.scala:73: no valid targets for annotation on value sc - it is discarded unused. You may specify targets with meta-annotations, e.g. (transient param)
[error] [warn]     transient sc: SparkContext,
```

This fix implemented here is the same as what we did in #8433: remove the `transient` annotations when they are not necessary and replace use  `transient private val` in the remaining cases.

Author: Josh Rosen <joshrosen@databricks.com>

Closes #10479 from JoshRosen/fix-sbt-2.11.
parent a6d38532
No related branches found
No related tags found
No related merge requests found
...@@ -70,26 +70,26 @@ class KinesisBackedBlockRDDPartition( ...@@ -70,26 +70,26 @@ class KinesisBackedBlockRDDPartition(
*/ */
private[kinesis] private[kinesis]
class KinesisBackedBlockRDD[T: ClassTag]( class KinesisBackedBlockRDD[T: ClassTag](
@transient sc: SparkContext, sc: SparkContext,
val regionName: String, val regionName: String,
val endpointUrl: String, val endpointUrl: String,
@transient blockIds: Array[BlockId], @transient private val _blockIds: Array[BlockId],
@transient val arrayOfseqNumberRanges: Array[SequenceNumberRanges], @transient val arrayOfseqNumberRanges: Array[SequenceNumberRanges],
@transient isBlockIdValid: Array[Boolean] = Array.empty, @transient private val isBlockIdValid: Array[Boolean] = Array.empty,
val retryTimeoutMs: Int = 10000, val retryTimeoutMs: Int = 10000,
val messageHandler: Record => T = KinesisUtils.defaultMessageHandler _, val messageHandler: Record => T = KinesisUtils.defaultMessageHandler _,
val awsCredentialsOption: Option[SerializableAWSCredentials] = None val awsCredentialsOption: Option[SerializableAWSCredentials] = None
) extends BlockRDD[T](sc, blockIds) { ) extends BlockRDD[T](sc, _blockIds) {
require(blockIds.length == arrayOfseqNumberRanges.length, require(_blockIds.length == arrayOfseqNumberRanges.length,
"Number of blockIds is not equal to the number of sequence number ranges") "Number of blockIds is not equal to the number of sequence number ranges")
override def isValid(): Boolean = true override def isValid(): Boolean = true
override def getPartitions: Array[Partition] = { override def getPartitions: Array[Partition] = {
Array.tabulate(blockIds.length) { i => Array.tabulate(_blockIds.length) { i =>
val isValid = if (isBlockIdValid.length == 0) true else isBlockIdValid(i) val isValid = if (isBlockIdValid.length == 0) true else isBlockIdValid(i)
new KinesisBackedBlockRDDPartition(i, blockIds(i), isValid, arrayOfseqNumberRanges(i)) new KinesisBackedBlockRDDPartition(i, _blockIds(i), isValid, arrayOfseqNumberRanges(i))
} }
} }
......
...@@ -30,7 +30,7 @@ import org.apache.spark.streaming.scheduler.ReceivedBlockInfo ...@@ -30,7 +30,7 @@ import org.apache.spark.streaming.scheduler.ReceivedBlockInfo
import org.apache.spark.streaming.{Duration, StreamingContext, Time} import org.apache.spark.streaming.{Duration, StreamingContext, Time}
private[kinesis] class KinesisInputDStream[T: ClassTag]( private[kinesis] class KinesisInputDStream[T: ClassTag](
@transient _ssc: StreamingContext, _ssc: StreamingContext,
streamName: String, streamName: String,
endpointUrl: String, endpointUrl: String,
regionName: String, regionName: String,
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment