java.lang.IncompatibleClassChangeError: In Apache Flink when checkpointing is enabled

I'm using Flink's KeyedProcessFunction to do some custom processing and I also use a value state to save my state. Sample code for the keyed process function and my state class is also given below. I get the following exception when I enable checkpointing. Is there a way to fix this? java.lang.IncompatibleClassChangeError: Class scala.collection.mutable.HashMap does not implement the requested interface scala.collection.mutable.SortedMap

2022-09-21 13:02:56,933 WARN  org.apache.flink.runtime.taskmanager.Task                    [] - KeyedProcess -> Sink: Unnamed (1/1)#6 (ff016f5bbf176977d893cecef4c158bb)
switched from RUNNING to FAILED with failure cause: java.lang.IncompatibleClassChangeError: Class scala.collection.mutable.HashMap does not implement the requested interface scala.collection.mutable.SortedMap
        at FlinkEventProcessor$Processor.$anonfun$processElement$2(FlinkEventProcessor.scala:1081)
        at FlinkEventProcessor$Processor.processElement(FlinkEventProcessor.scala:1079)
        at FlinkEventProcessor$Processor.processElement(FlinkEventProcessor.scala:941)
        at org.apache.flink.streaming.api.operators.KeyedProcessOperator.processElement(KeyedProcessOperator.java:83)
        at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:205)
        at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:134)
        at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:105)
        at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:66)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:423)
        at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:204)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:681)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.executeInvoke(StreamTask.java:636)
       at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:647)
        at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:620)
        at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:779)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)
        at java.base/java.lang.Thread.run(Thread.java:829)

Sample code

case class State(key: String, var lastSequenceNo: Long) {
  val outOfOrder: mutable.SortedMap[Long, KafkaMessage] = mutable.SortedMap.empty[Long, KafkaMessage]
}


class Processor()
  extends KeyedProcessFunction[String, KafkaMessage, Row] {
    private var valueState: ValueState[State] = _

    override def open(parameters: Configuration): Unit = {
        valueState = getRuntimeContext.getState(new ValueStateDescriptor("state", classOf[State]))
    }

    override def processElement(in: KafkaMessage,
                            ctx: KeyedProcessFunction[String, KafkaMessage, Row]#Context,
                            out: Collector[Row]): Unit = {

    logger.debug("Processing out of order messages...")
    while (state.outOfOrder.nonEmpty) {
      logger.debug("Processing out of order message")
      val min = state.outOfOrder.keys.head
      state.outOfOrder.remove(min)
    }
  }
}