Re: Null pointer exception while replying WAL
OK
Getting Null pointer exception while replying WAL! One possible reason is
that the messages RDD might contain null elements, and attempting to read
JSON from null values can result in an NPE. To handle this, you can add a
filter before processing the RDD to remove null elements.
msgs.foreachRDD { rdd =>
if (rdd.take(1).nonEmpty) {
val messages: RDD[String] = rdd
.map { sr =>
Option(sr).getOrElse("NO records found")
}
.filter(_ != "NO records found")
try {
val messagesJson = spark.read.json(messages)
messagesJson.write.mode("append").parquet(data)
} catch {
case ex: Exception =>
ex.printStackTrace()
}
}
}
This modification uses Option *t*o handle potential null values in the rdd
and filters out any elements that are still "NO records found" after the
mapping operation.
HTH
Mich Talebzadeh,
Dad | Technologist | Solutions Architect | Engineer
London
United Kingdom
view my Linkedin profile
>
>
>
> On Mon, Feb 12, 2024 at 4:23 AM Mich Talebzadeh
> wrote:
>
>> Hi,
>>
>> It is challenging to make a recommendation without further details. I am
>> guessing you are trying to build a fault-tolerant spark application (spark
>> structured streaming) that consumes messages from Solace?
>> To address *NullPointerException* in the
Re: Null pointer exception while replying WAL
Please find below code
def main(args: Array[String]): Unit = {
val config: Config = ConfigFactory.load()
val streamC = StreamingContext.getOrCreate(
checkpointDirectory,
() => functionToCreateContext(config, checkpointDirectory)
)
streamC.start()
streamC.awaitTermination()
}
def functionToCreateContext(config: Config, checkpointDirectory: String):
StreamingContext = {
val brokerUrl = config.getString("streaming.solace.brokerURL")
val username = config.getString("streaming.solace.userName")
val passwordSol = config.getString("streaming.solace.password")
val vpn = config.getString("streaming.solace.vpn")
val queue = config.getString("streaming.solace.queueName")
val connectionFactory =
config.getString("streaming.solace.connectionFactory")
val spark = SparkSession
.builder()
.appName("rem-Streaming-Consumer")
.config("spark.streaming.receiver.writeAheadLog.enable", "true")
.config("spark.streaming.blockInterval", blockInterval)
.config("spark.serializer",
"org.apache.spark.serializer.KryoSerializer")
.config("spark.streaming.receiver.writeAheadLog.enable", "true")
.enableHiveSupport
.getOrCreate()
val sc = spark.sparkContext
val ssc = new StreamingContext(sc, Seconds(batchInterval))
ssc.checkpoint(checkpointDirectory)
val converter: Message => Option[String] = {
case msg: TextMessage =>
Some(msg.getText)
case _ =>
None
}
val props = new Properties()
props.setProperty(
Context.INITIAL_CONTEXT_FACTORY,
"com.solacesystems.jndi.SolJNDIInitialContextFactory"
)
props.setProperty(Context.PROVIDER_URL, brokerUrl)
props.setProperty(Context.SECURITY_PRINCIPAL, username)
props.setProperty(Context.SECURITY_PRINCIPAL, passwordSol)
props.setProperty(SupportedProperty.SOLACE_JMS_VPN, vpn)
val msgs = JmsStreamUtils.createSynchronousJmsQueueStream(
ssc,
JndiMessageConsumerFactory(props,QueueJmsDestinationInfo(queue),
connectionFactoryName
= connectionFactory,messageSelector =
""),converter,1000, 1.second,10.second,StorageLevel.MEMORY_AND_DISK_SER_2 )
msgs.foreachRDD(rdd =>
if (rdd.take(1).length > 0) {
val messages: RDD[String] = rdd.map { sr =>
if (sr == null) {
println("NO records found")
"NO records found"
} else {
println("Input Records from Solace queue : " + sr.toString)
sr.toString
}
}
Thread.sleep(12)
try{
* val messagesJson = spark.read.json(messages) ===> getting NPE here
after restarting using WAL*
messagesJson.write.mode("append").parquet(data)
}
catch {
case ex => ex.printStackTrace()
}
})
ssc
}
Thanks & Regards,
Nayan Sharma
*+91-8095382952*
On Mon, Feb 12, 2024 at 4:23 AM Mich Talebzadeh
wrote:
> Hi,
>
> It is challenging to make a recommendation without further details. I am
> guessing you are trying to build a fault-tolerant spark application (spark
> structured streaming) that consumes messages from Solace?
> To address *NullPointerException* in the context of the provided
> information, you need to review the part of the code where the exception is
> thrown and identifying which object or method call is resulting in *null* can
> help the debugging process plus checking the logs.
>
> HTH
>
> Mich Talebzadeh,
> Dad | Technologist | Solutions Architect | Engineer
> London
> United Kingdom
>
>
>view my Linkedin profile
>
