Exceptions when using scala types directly with the State Process API

// code placeholder
  class ReaderFunction extends KeyedStateReaderFunction[Long, (Long, Long)] {
    var countState: ValueState[(Long, Long)] = _
    override def open(parameters: Configuration): Unit = {
      val stateDescriptor = new ValueStateDescriptor("average", createTypeInformation[(Long, Long)])
      countState = getRuntimeContext().getState(stateDescriptor)
    }    override def readKey(key: Long, ctx: KeyedStateReaderFunction.Context, out: Collector[(Long, Long)]): Unit = {
      out.collect(countState.value())
    }
  }