i'm having hard time trying figure out reason keep seeing:
`hibernateoptimisticlockingfailureexception: flowexecution: optimistic locking failed; nested exception org.hibernate.staleobjectstateexception: row updated or deleted transaction (or unsaved-value mapping incorrect)` i have service uses quartz scheduler fire jobs, in context these jobs called flows , each flow might composed several tasks, flows , tasks executables , info actual executions stored flowexecutions , taskexecutions. service uses flowservice start flows.
upd: there's quartz job, "executorjob" responsible firing flows/tasks. when it's triggered uses flowservice start whatever supposed to. i'm wondering if it's possible quartz thread not creating new hibernate sessions each time uses service , that's cause of problem. haven't changed scope of flowservice it's singleton, how gorm manages session used it?
upd2: tried using persistencecontextinterceptor on executorjob make sure each use of service uses new session didn't solved problem. simplified code executorjob added.
i couldn't reproduce issue locally, it's happening on production, more when there's lot of flows initiate. i've tried synchronizing execute methods tasks , flows didn't worked, i'll try use pessimistic lock guess won't solve issue since checking application logs seems there aren't 2 threads updating same row. following i've tried show simplified version of code mimicking how project structured.
// ------------------ // domain classes // ------------------ abstract class executable { static hasmany = [flowtasks: flowtask] static transients = ['executions'] list<execution> getexecutions() { this.id ? execution.findallbyexecutable(this) : [] } void addtoexecutions(execution execution) { execution.executable = execution.save() } abstract list<execution> execute(map params) } class flow extends executable { sortedset<flowtask> tasks static hasmany = [tasks: flowtask] private static final object lockexecute = new object() private static final object lockexecutetask = new object() list<flowexecution> execute(map params) { synchronized (lockexecute) { list<map> multiparams = multiplyparams(params) multiparams.collect { map param -> flowexecution flowexecution = new flowexecution() addtoexecutions(flowexecution) flowexecution.save() this.attach() save() executetasks(firsttasks(param), flowexecution, param) } } } list<map> multiplyparams(map params) { // creates list of params executions must started [params] } set<flowtask> firsttasks(map params) { // finds first tasks executed flow tasks.findadll { true } } private flowexecution executetasks(set<flowtask> tasks, flowexecution flowexecution, map params) { synchronized (lockexecutetask) { tasks.each { flowtask flowtask -> try { list<execution> executions = flowtask.execute(params) executions.each { execution execution -> flowexecution.addtoexecutions(execution) } flowexecution.attach() } catch { // log error executing task throw e } } this.attach() try { save(flush: true) } catch (hibernateoptimisticlockingfailureexception e) { // log error saving flow throw e } flowexecution } } } class task extends executable { private static final object lockexecute = new object() private static final object lockgetexecution = new object() taskexecution execute(taskexecution execution) { taskservice.start(execution) execution } list<taskexecution> execute(map params) { synchronized (lockexecute) { list<map> multiexecparams = multiplyparams(params) multiexecparams.collect { map param -> taskexecution execution = getexecution(param) execute(execution) } } } taskexecution getexecution(map params) { synchronized (lockgetexecution) { taskexecution execution = new taskexecution(executable: this) execution.setparameters(params) addtoexecutions(execution) execution.attach() execution.flowexecution?.attach() this.attach() try { save(flush: true) } catch (hibernateoptimisticlockingfailureexception e) { // log error saving task throw e } execution } } list<map> multiplyparams(map params) { // creates list of params tasks must started [params] } } class flowtask { static belongsto = [flow: flow, executable: executable] list<execution> execute(map params) { executable.execute(params) } } abstract class execution { map parameterdata = [:] static belongsto = [executable: executable, flowexecution: flowexecution] static transients = ['parameters', 'taskexecutions'] void setparameters(map params) { params.each { key, value -> parameterdata[key] = jsonparser.tojson(value) } } } class taskexecution extends execution { } class flowexecution extends execution { list<execution> executions static transients = ['executions'] flowexecution() { executions = [] } set<taskexecution> gettaskexecutions() { executions?.collect { execution execution -> return execution.taskexecution }?.flatten()?.toset() } void addtoexecutions(execution execution){ executions.add(execution) execution.flowexecution = execution.save() } def onload() { try { executions = this.id ? execution.findallbyflowexecution(this) : [] } catch (exception e){ log.error(e) [] } } } // ----------------- // service classes // ----------------- class flowservice { map start(long flowid, map params) { flow flow = flow.lock(flowid) startflow(flow, params) } private map startflow(flow flow, map params) { list<runningflow> runningflows = flow.execute(params) [data: [success: true], status: http_ok] } } //-------------------------------------- // quartz job //-------------------------------------- class executorjob implements interruptablejob { def grailsapplication = holders.getgrailsapplication() static triggers = {} private thread thread void execute(jobexecutioncontext context) throws jobexecutionexception { thread = thread.currentthread() synchronized (lockcontainer.tasklock) { map params = context.mergedjobdatamap def persistenceinterceptor = persistenceinterceptorinstance try { persistenceinterceptor.init() long executableid = params.executableid long def service = (executable.get(executableid) instanceof flow) ? flowserviceinstance : taskserviceinstance service.start(executableid, params) } catch (exception e) { // log error } { persistenceinterceptor.flush() persistenceinterceptor.destroy() } } } persistencecontextinterceptor getpersistenceinterceptorinstance() { grailsapplication.maincontext.getbean('persistenceinterceptor') } fluxoservice getflowserviceinstance() { grailsapplication.maincontext.getbean('flowservice') } tarefaservice gettaskserviceinstance() { grailsapplication.maincontext.getbean('taskservice') } @override void interrupt() throws unabletointerruptjobexception { thread?.interrupt() } } anyone knows might help?
well, it's hard understand going wrong. however, guess error gets thrown when have object in session has been saved or updated other transaction. again, when hibernate tries save object gives row updated error transaction error.
i guess can try refresh before save object , see how goes.
http://grails.github.io/grails-doc/2.3.4/ref/domain%20classes/refresh.html
def b = book.get(1) … b.refresh() 
Comments
Post a Comment