Github user yssharma commented on the issue:
https://github.com/apache/spark/pull/17467
@brkyvz - Added new changes that adds -
- A case class `KinesisReadConfigurations` that adds all the kinesis read
configs in a single place
- A test class that passes the kinesis configs in `SparkConf` which are
then used to create the kinesis configs object in `KinesisInputDStream` and
passed down to `KinesisBackedBlockRDD`
- Docs improvement
I also played with the `PrivateMethodTester ` but wasn't able to access the
private function `KinesisSequenceRangeIterator#retryOrTimeout` . Probably
because of the generics used in the function. I used an alternative to fetch
the RDD's directly and check the configs passed in there.
I would still like to learn how to get the `retryOrTimeout` working just
out of interest. Adding the error below:
```
// KinesisSequenceRangeIterator # retryOrTimeout
val retryOrTimeoutMethod = PrivateMethod[Object]('retryOrTimeout) //
<<<- Issue
val partitions = kinesisRDD.partitions.map {
_.asInstanceOf[KinesisBackedBlockRDDPartition] }.toSeq
seqNumRanges1.ranges.map{ range =>
val seqRangeIter =
new
KinesisSequenceRangeIterator(DefaultCredentials.provider.getCredentials,
dummyEndpointUrl, dummyRegionName, range,
kinesisRDD.kinesisReadConfigs)
seqRangeIter.invokePrivate(retryOrTimeoutMethod("Passing custom
message"))
}
- Kinesis read with custom configurations *** FAILED ***
java.lang.IllegalArgumentException: Can't find a private method named:
retryOrTimeout
at
org.scalatest.PrivateMethodTester$Invoker.invokePrivate(PrivateMethodTester.scala:247)
at
org.apache.spark.streaming.kinesis.KinesisStreamTests$$anonfun$7$$anonfun$apply$mcV$sp$13.apply(KinesisStreamSuite.scala:286)
at
org.apache.spark.streaming.kinesis.KinesisStreamTests$$anonfun$7$$anonfun$apply$mcV$sp$13.apply(KinesisStreamSuite.scala:281)
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at
scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.immutable.List.foreach(List.scala:381)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.immutable.List.map(List.scala:285)
at
org.apache.spark.streaming.kinesis.KinesisStreamTests$$anonfun$7.apply$mcV$sp(KinesisStreamSuite.scala:281)
at
org.apache.spark.streaming.kinesis.KinesisStreamTests$$anonfun$7.apply(KinesisStreamSuite.scala:237)
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
