Cassandra RDD not getting refreshed with new rows in Spark Streaming

[sonyjv]

Hi,

I have the following Spark streaming application which stream from a Kafka topic, do some processing and publish the result to another topic. In between I am using Calliope to read rows from a Cassandra CF.

The issue is when the application is running, if a new row is inserted into Cassandra CF, that new row will not be available to the application until i stop and restart the application.

def main(args: scala.Array[String]) {
..................
val ssc = new StreamingContext(conf, Milliseconds(2000))
val searches = KafkaUtils.createStream(ssc, zkQuorum,group, topicMap).map(line => parse(line._2).extract[Search])
.....................

val casOffer = CasBuilder.cql3.withColumnFamily("optimiti", "offer").onHost(cassandra_ip).onPort(cassandra_port);
var cqlOffers = ssc.sc.cql3CassandraOffer

val result = ..........

result.foreachRDD( res =>{
val resStr = res.collect;
if(!resStr.isEmpty){
for (i <- 0 to resStr.length-1) {
producer.send(new KeyedMessage[String, String](topics_topush, ""+ resStr%28i)))
}
}
})
ssc.start()
ssc.awaitTermination()
}

I had tried to cache cqlOffers and unpersist it once the result is calculated without success.Is there anything that I am missing. Is it a problem with Calliope?

Thanks,
Sony