@Test public void testRetryTaskRun() throws Throwable { new TransactionManager().new RetryTask().run(); assertTrue(true, "Test completed without Exception"); }
@Override public void run () { long id = 0; int session = 0; List<TransactionParticipant> members = null; Iterator<TransactionParticipant> iter = null; PausedTransaction pt; boolean abort = false; LogEvent evt = null; Profiler prof; boolean paused; boolean transactionActive; Thread thread = Thread.currentThread(); if (threads.size() < maxSessions) { threads.add(thread); session = threads.indexOf(thread); activeSessions.incrementAndGet(); } else { getLog().warn ("Max sessions reached, new session not created"); return; } getLog().info ("start " + thread); while (running()) { Serializable context = null; prof = null; evt = null; paused = false; transactionActive = false; thread.setName (getName() + "-" + session + ":idle"); int action = -1; try { if (hasStatusListeners) notifyStatusListeners (session, TransactionStatusEvent.State.READY, id, "", null); Object obj = iisp.in (queue, MAX_WAIT); if (obj == Boolean.FALSE) continue; if (obj == null) { if (session+1 > sessions && getActiveSessions() > sessions) break; else { continue; } } if (!(obj instanceof Serializable)) { getLog().error ( "non serializable '" + obj.getClass().getName() + "' on queue '" + queue + "'" ); continue; } context = (Serializable) obj; if (obj instanceof Pausable) { Pausable pausable = (Pausable) obj; pt = pausable.getPausedTransaction(); if (pt != null) { pt.cancelExpirationMonitor(); id = pt.id(); members = pt.members(); iter = pt.iterator(); abort = pt.isAborting(); evt = pt.getLogEvent(); prof = pt.getProfiler(); if (metrics != null && prof != null) metrics.record(getName(pt.getParticipant()) + "-resume", prof.getPartialInMillis()); if (prof != null) prof.reenable(); pausedCounter.decrementAndGet(); } } else pt = null; if (pt == null) { int running = getActiveTransactions(); if (maxActiveSessions > 0 && running >= maxActiveSessions) { getLog().warn ( Thread.currentThread().getName() + ": emergency retry, running-sessions=" + running + ", max-active-sessions=" + maxActiveSessions ); psp.out (RETRY_QUEUE, obj, retryTimeout); checkRetryTask(); continue; } abort = false; id = nextId (); members = new ArrayList (); iter = getParticipants (DEFAULT_GROUP).iterator(); activeTransactions.incrementAndGet(); } transactionActive = true; if (debug) { if (evt == null) { evt = getLog().createLogEvent("debug", Thread.currentThread().getName() + ":" + Long.toString(id) + (pt != null ? " [resuming]" : "") ); if (debugContext) { evt.addMessage (context); } } if (prof == null) prof = new Profiler(); else prof.checkPoint("resume"); } snapshot (id, context, PREPARING); setThreadLocal(id, context); action = prepare (session, id, context, members, iter, abort, evt, prof); removeThreadLocal(); switch (action) { case PAUSE: paused = true; if (id % TIMER_PURGE_INTERVAL == 0) timer.purge(); pausedCounter.incrementAndGet(); break; case PREPARED: if (members.size() > 0) { setState(id, COMMITTING); setThreadLocal(id, context); commit(session, id, context, members, false, evt, prof); removeThreadLocal(); } break; case ABORTED: if (members.size() > 0) { setThreadLocal(id, context); abort(session, id, context, members, false, evt, prof); removeThreadLocal(); } break; case RETRY: psp.out (RETRY_QUEUE, context); checkRetryTask(); break; case NO_JOIN: break; } if ((action & PAUSE) == 0) { snapshot (id, null, DONE); if (id == tail) { checkTail (); } else { purge (id, false); } tps.tick(); } } catch (Throwable t) { if (evt == null) getLog().fatal (t); else evt.addMessage (t); } finally { removeThreadLocal(); if (transactionActive && !paused) activeTransactions.decrementAndGet(); if (hasStatusListeners) { notifyStatusListeners ( session, paused ? TransactionStatusEvent.State.PAUSED : TransactionStatusEvent.State.DONE, id, "", context); } if (evt != null && (action == PREPARED || action == ABORTED || (action == -1 && prof != null))) { switch (action) { case PREPARED : evt.setTag("commit"); break; case ABORTED : evt.setTag ("abort"); break; case -1: evt.setTag ("undefined"); break; } if (getInTransit() > Math.max(maxActiveSessions, activeSessions.get()) * 100) { evt.addMessage("WARNING: IN-TRANSIT TOO HIGH"); } evt.addMessage ( String.format (" %s, elapsed=%dms", tmInfo(), prof != null ? prof.getElapsedInMillis() : -1 ) ); if (prof != null) evt.addMessage (prof); try { Logger.log(freeze(context, evt, prof)); } catch (Throwable t) { getLog().error(t); } } } } threads.remove(thread); int currentActiveSessions = activeSessions.decrementAndGet(); getLog().info ("stop " + Thread.currentThread() + ", active sessions=" + currentActiveSessions); }
TransactionManager extends QBeanSupport implements Runnable, TransactionConstants, TransactionManagerMBean, Loggeable, MetricsProvider { @Override public void run () { long id = 0; int session = 0; List<TransactionParticipant> members = null; Iterator<TransactionParticipant> iter = null; PausedTransaction pt; boolean abort = false; LogEvent evt = null; Profiler prof; boolean paused; boolean transactionActive; Thread thread = Thread.currentThread(); if (threads.size() < maxSessions) { threads.add(thread); session = threads.indexOf(thread); activeSessions.incrementAndGet(); } else { getLog().warn ("Max sessions reached, new session not created"); return; } getLog().info ("start " + thread); while (running()) { Serializable context = null; prof = null; evt = null; paused = false; transactionActive = false; thread.setName (getName() + "-" + session + ":idle"); int action = -1; try { if (hasStatusListeners) notifyStatusListeners (session, TransactionStatusEvent.State.READY, id, "", null); Object obj = iisp.in (queue, MAX_WAIT); if (obj == Boolean.FALSE) continue; if (obj == null) { if (session+1 > sessions && getActiveSessions() > sessions) break; else { continue; } } if (!(obj instanceof Serializable)) { getLog().error ( "non serializable '" + obj.getClass().getName() + "' on queue '" + queue + "'" ); continue; } context = (Serializable) obj; if (obj instanceof Pausable) { Pausable pausable = (Pausable) obj; pt = pausable.getPausedTransaction(); if (pt != null) { pt.cancelExpirationMonitor(); id = pt.id(); members = pt.members(); iter = pt.iterator(); abort = pt.isAborting(); evt = pt.getLogEvent(); prof = pt.getProfiler(); if (metrics != null && prof != null) metrics.record(getName(pt.getParticipant()) + "-resume", prof.getPartialInMillis()); if (prof != null) prof.reenable(); pausedCounter.decrementAndGet(); } } else pt = null; if (pt == null) { int running = getActiveTransactions(); if (maxActiveSessions > 0 && running >= maxActiveSessions) { getLog().warn ( Thread.currentThread().getName() + ": emergency retry, running-sessions=" + running + ", max-active-sessions=" + maxActiveSessions ); psp.out (RETRY_QUEUE, obj, retryTimeout); checkRetryTask(); continue; } abort = false; id = nextId (); members = new ArrayList (); iter = getParticipants (DEFAULT_GROUP).iterator(); activeTransactions.incrementAndGet(); } transactionActive = true; if (debug) { if (evt == null) { evt = getLog().createLogEvent("debug", Thread.currentThread().getName() + ":" + Long.toString(id) + (pt != null ? " [resuming]" : "") ); if (debugContext) { evt.addMessage (context); } } if (prof == null) prof = new Profiler(); else prof.checkPoint("resume"); } snapshot (id, context, PREPARING); setThreadLocal(id, context); action = prepare (session, id, context, members, iter, abort, evt, prof); removeThreadLocal(); switch (action) { case PAUSE: paused = true; if (id % TIMER_PURGE_INTERVAL == 0) timer.purge(); pausedCounter.incrementAndGet(); break; case PREPARED: if (members.size() > 0) { setState(id, COMMITTING); setThreadLocal(id, context); commit(session, id, context, members, false, evt, prof); removeThreadLocal(); } break; case ABORTED: if (members.size() > 0) { setThreadLocal(id, context); abort(session, id, context, members, false, evt, prof); removeThreadLocal(); } break; case RETRY: psp.out (RETRY_QUEUE, context); checkRetryTask(); break; case NO_JOIN: break; } if ((action & PAUSE) == 0) { snapshot (id, null, DONE); if (id == tail) { checkTail (); } else { purge (id, false); } tps.tick(); } } catch (Throwable t) { if (evt == null) getLog().fatal (t); else evt.addMessage (t); } finally { removeThreadLocal(); if (transactionActive && !paused) activeTransactions.decrementAndGet(); if (hasStatusListeners) { notifyStatusListeners ( session, paused ? TransactionStatusEvent.State.PAUSED : TransactionStatusEvent.State.DONE, id, "", context); } if (evt != null && (action == PREPARED || action == ABORTED || (action == -1 && prof != null))) { switch (action) { case PREPARED : evt.setTag("commit"); break; case ABORTED : evt.setTag ("abort"); break; case -1: evt.setTag ("undefined"); break; } if (getInTransit() > Math.max(maxActiveSessions, activeSessions.get()) * 100) { evt.addMessage("WARNING: IN-TRANSIT TOO HIGH"); } evt.addMessage ( String.format (" %s, elapsed=%dms", tmInfo(), prof != null ? prof.getElapsedInMillis() : -1 ) ); if (prof != null) evt.addMessage (prof); try { Logger.log(freeze(context, evt, prof)); } catch (Throwable t) { getLog().error(t); } } } } threads.remove(thread); int currentActiveSessions = activeSessions.decrementAndGet(); getLog().info ("stop " + Thread.currentThread() + ", active sessions=" + currentActiveSessions); } }
TransactionManager extends QBeanSupport implements Runnable, TransactionConstants, TransactionManagerMBean, Loggeable, MetricsProvider { @Override public void run () { long id = 0; int session = 0; List<TransactionParticipant> members = null; Iterator<TransactionParticipant> iter = null; PausedTransaction pt; boolean abort = false; LogEvent evt = null; Profiler prof; boolean paused; boolean transactionActive; Thread thread = Thread.currentThread(); if (threads.size() < maxSessions) { threads.add(thread); session = threads.indexOf(thread); activeSessions.incrementAndGet(); } else { getLog().warn ("Max sessions reached, new session not created"); return; } getLog().info ("start " + thread); while (running()) { Serializable context = null; prof = null; evt = null; paused = false; transactionActive = false; thread.setName (getName() + "-" + session + ":idle"); int action = -1; try { if (hasStatusListeners) notifyStatusListeners (session, TransactionStatusEvent.State.READY, id, "", null); Object obj = iisp.in (queue, MAX_WAIT); if (obj == Boolean.FALSE) continue; if (obj == null) { if (session+1 > sessions && getActiveSessions() > sessions) break; else { continue; } } if (!(obj instanceof Serializable)) { getLog().error ( "non serializable '" + obj.getClass().getName() + "' on queue '" + queue + "'" ); continue; } context = (Serializable) obj; if (obj instanceof Pausable) { Pausable pausable = (Pausable) obj; pt = pausable.getPausedTransaction(); if (pt != null) { pt.cancelExpirationMonitor(); id = pt.id(); members = pt.members(); iter = pt.iterator(); abort = pt.isAborting(); evt = pt.getLogEvent(); prof = pt.getProfiler(); if (metrics != null && prof != null) metrics.record(getName(pt.getParticipant()) + "-resume", prof.getPartialInMillis()); if (prof != null) prof.reenable(); pausedCounter.decrementAndGet(); } } else pt = null; if (pt == null) { int running = getActiveTransactions(); if (maxActiveSessions > 0 && running >= maxActiveSessions) { getLog().warn ( Thread.currentThread().getName() + ": emergency retry, running-sessions=" + running + ", max-active-sessions=" + maxActiveSessions ); psp.out (RETRY_QUEUE, obj, retryTimeout); checkRetryTask(); continue; } abort = false; id = nextId (); members = new ArrayList (); iter = getParticipants (DEFAULT_GROUP).iterator(); activeTransactions.incrementAndGet(); } transactionActive = true; if (debug) { if (evt == null) { evt = getLog().createLogEvent("debug", Thread.currentThread().getName() + ":" + Long.toString(id) + (pt != null ? " [resuming]" : "") ); if (debugContext) { evt.addMessage (context); } } if (prof == null) prof = new Profiler(); else prof.checkPoint("resume"); } snapshot (id, context, PREPARING); setThreadLocal(id, context); action = prepare (session, id, context, members, iter, abort, evt, prof); removeThreadLocal(); switch (action) { case PAUSE: paused = true; if (id % TIMER_PURGE_INTERVAL == 0) timer.purge(); pausedCounter.incrementAndGet(); break; case PREPARED: if (members.size() > 0) { setState(id, COMMITTING); setThreadLocal(id, context); commit(session, id, context, members, false, evt, prof); removeThreadLocal(); } break; case ABORTED: if (members.size() > 0) { setThreadLocal(id, context); abort(session, id, context, members, false, evt, prof); removeThreadLocal(); } break; case RETRY: psp.out (RETRY_QUEUE, context); checkRetryTask(); break; case NO_JOIN: break; } if ((action & PAUSE) == 0) { snapshot (id, null, DONE); if (id == tail) { checkTail (); } else { purge (id, false); } tps.tick(); } } catch (Throwable t) { if (evt == null) getLog().fatal (t); else evt.addMessage (t); } finally { removeThreadLocal(); if (transactionActive && !paused) activeTransactions.decrementAndGet(); if (hasStatusListeners) { notifyStatusListeners ( session, paused ? TransactionStatusEvent.State.PAUSED : TransactionStatusEvent.State.DONE, id, "", context); } if (evt != null && (action == PREPARED || action == ABORTED || (action == -1 && prof != null))) { switch (action) { case PREPARED : evt.setTag("commit"); break; case ABORTED : evt.setTag ("abort"); break; case -1: evt.setTag ("undefined"); break; } if (getInTransit() > Math.max(maxActiveSessions, activeSessions.get()) * 100) { evt.addMessage("WARNING: IN-TRANSIT TOO HIGH"); } evt.addMessage ( String.format (" %s, elapsed=%dms", tmInfo(), prof != null ? prof.getElapsedInMillis() : -1 ) ); if (prof != null) evt.addMessage (prof); try { Logger.log(freeze(context, evt, prof)); } catch (Throwable t) { getLog().error(t); } } } } threads.remove(thread); int currentActiveSessions = activeSessions.decrementAndGet(); getLog().info ("stop " + Thread.currentThread() + ", active sessions=" + currentActiveSessions); }  }
TransactionManager extends QBeanSupport implements Runnable, TransactionConstants, TransactionManagerMBean, Loggeable, MetricsProvider { @Override public void run () { long id = 0; int session = 0; List<TransactionParticipant> members = null; Iterator<TransactionParticipant> iter = null; PausedTransaction pt; boolean abort = false; LogEvent evt = null; Profiler prof; boolean paused; boolean transactionActive; Thread thread = Thread.currentThread(); if (threads.size() < maxSessions) { threads.add(thread); session = threads.indexOf(thread); activeSessions.incrementAndGet(); } else { getLog().warn ("Max sessions reached, new session not created"); return; } getLog().info ("start " + thread); while (running()) { Serializable context = null; prof = null; evt = null; paused = false; transactionActive = false; thread.setName (getName() + "-" + session + ":idle"); int action = -1; try { if (hasStatusListeners) notifyStatusListeners (session, TransactionStatusEvent.State.READY, id, "", null); Object obj = iisp.in (queue, MAX_WAIT); if (obj == Boolean.FALSE) continue; if (obj == null) { if (session+1 > sessions && getActiveSessions() > sessions) break; else { continue; } } if (!(obj instanceof Serializable)) { getLog().error ( "non serializable '" + obj.getClass().getName() + "' on queue '" + queue + "'" ); continue; } context = (Serializable) obj; if (obj instanceof Pausable) { Pausable pausable = (Pausable) obj; pt = pausable.getPausedTransaction(); if (pt != null) { pt.cancelExpirationMonitor(); id = pt.id(); members = pt.members(); iter = pt.iterator(); abort = pt.isAborting(); evt = pt.getLogEvent(); prof = pt.getProfiler(); if (metrics != null && prof != null) metrics.record(getName(pt.getParticipant()) + "-resume", prof.getPartialInMillis()); if (prof != null) prof.reenable(); pausedCounter.decrementAndGet(); } } else pt = null; if (pt == null) { int running = getActiveTransactions(); if (maxActiveSessions > 0 && running >= maxActiveSessions) { getLog().warn ( Thread.currentThread().getName() + ": emergency retry, running-sessions=" + running + ", max-active-sessions=" + maxActiveSessions ); psp.out (RETRY_QUEUE, obj, retryTimeout); checkRetryTask(); continue; } abort = false; id = nextId (); members = new ArrayList (); iter = getParticipants (DEFAULT_GROUP).iterator(); activeTransactions.incrementAndGet(); } transactionActive = true; if (debug) { if (evt == null) { evt = getLog().createLogEvent("debug", Thread.currentThread().getName() + ":" + Long.toString(id) + (pt != null ? " [resuming]" : "") ); if (debugContext) { evt.addMessage (context); } } if (prof == null) prof = new Profiler(); else prof.checkPoint("resume"); } snapshot (id, context, PREPARING); setThreadLocal(id, context); action = prepare (session, id, context, members, iter, abort, evt, prof); removeThreadLocal(); switch (action) { case PAUSE: paused = true; if (id % TIMER_PURGE_INTERVAL == 0) timer.purge(); pausedCounter.incrementAndGet(); break; case PREPARED: if (members.size() > 0) { setState(id, COMMITTING); setThreadLocal(id, context); commit(session, id, context, members, false, evt, prof); removeThreadLocal(); } break; case ABORTED: if (members.size() > 0) { setThreadLocal(id, context); abort(session, id, context, members, false, evt, prof); removeThreadLocal(); } break; case RETRY: psp.out (RETRY_QUEUE, context); checkRetryTask(); break; case NO_JOIN: break; } if ((action & PAUSE) == 0) { snapshot (id, null, DONE); if (id == tail) { checkTail (); } else { purge (id, false); } tps.tick(); } } catch (Throwable t) { if (evt == null) getLog().fatal (t); else evt.addMessage (t); } finally { removeThreadLocal(); if (transactionActive && !paused) activeTransactions.decrementAndGet(); if (hasStatusListeners) { notifyStatusListeners ( session, paused ? TransactionStatusEvent.State.PAUSED : TransactionStatusEvent.State.DONE, id, "", context); } if (evt != null && (action == PREPARED || action == ABORTED || (action == -1 && prof != null))) { switch (action) { case PREPARED : evt.setTag("commit"); break; case ABORTED : evt.setTag ("abort"); break; case -1: evt.setTag ("undefined"); break; } if (getInTransit() > Math.max(maxActiveSessions, activeSessions.get()) * 100) { evt.addMessage("WARNING: IN-TRANSIT TOO HIGH"); } evt.addMessage ( String.format (" %s, elapsed=%dms", tmInfo(), prof != null ? prof.getElapsedInMillis() : -1 ) ); if (prof != null) evt.addMessage (prof); try { Logger.log(freeze(context, evt, prof)); } catch (Throwable t) { getLog().error(t); } } } } threads.remove(thread); int currentActiveSessions = activeSessions.decrementAndGet(); getLog().info ("stop " + Thread.currentThread() + ", active sessions=" + currentActiveSessions); }  @Override void initService(); @Override void startService(); @Override void stopService(); void queue(Serializable context); void push(Serializable context); @SuppressWarnings("unused") String getQueueName(); Space getSpace(); Space getInputSpace(); Space getPersistentSpace(); @Override void run(); @Override long getTail(); @Override long getHead(); long getInTransit(); @Override void setConfiguration(Configuration cfg); void addListener(TransactionStatusListener l); void removeListener(TransactionStatusListener l); TPS getTPS(); @Override String getTPSAsString(); @Override float getTPSAvg(); @Override int getTPSPeak(); @Override Date getTPSPeakWhen(); @Override long getTPSElapsed(); @Override void resetTPS(); @Override Metrics getMetrics(); @Override void dump(PrintStream ps, String indent); TransactionParticipant createParticipant(Element e); @Override int getOutstandingTransactions(); @Override void setDebug(boolean debug); @Override boolean getDebugContext(); @Override void setDebugContext(boolean debugContext); @Override boolean getDebug(); @Override int getActiveSessions(); int getPausedCounter(); int getActiveTransactions(); int getMaxSessions(); static Serializable getSerializable(); static T getContext(); static Long getId(); }
TransactionManager extends QBeanSupport implements Runnable, TransactionConstants, TransactionManagerMBean, Loggeable, MetricsProvider { @Override public void run () { long id = 0; int session = 0; List<TransactionParticipant> members = null; Iterator<TransactionParticipant> iter = null; PausedTransaction pt; boolean abort = false; LogEvent evt = null; Profiler prof; boolean paused; boolean transactionActive; Thread thread = Thread.currentThread(); if (threads.size() < maxSessions) { threads.add(thread); session = threads.indexOf(thread); activeSessions.incrementAndGet(); } else { getLog().warn ("Max sessions reached, new session not created"); return; } getLog().info ("start " + thread); while (running()) { Serializable context = null; prof = null; evt = null; paused = false; transactionActive = false; thread.setName (getName() + "-" + session + ":idle"); int action = -1; try { if (hasStatusListeners) notifyStatusListeners (session, TransactionStatusEvent.State.READY, id, "", null); Object obj = iisp.in (queue, MAX_WAIT); if (obj == Boolean.FALSE) continue; if (obj == null) { if (session+1 > sessions && getActiveSessions() > sessions) break; else { continue; } } if (!(obj instanceof Serializable)) { getLog().error ( "non serializable '" + obj.getClass().getName() + "' on queue '" + queue + "'" ); continue; } context = (Serializable) obj; if (obj instanceof Pausable) { Pausable pausable = (Pausable) obj; pt = pausable.getPausedTransaction(); if (pt != null) { pt.cancelExpirationMonitor(); id = pt.id(); members = pt.members(); iter = pt.iterator(); abort = pt.isAborting(); evt = pt.getLogEvent(); prof = pt.getProfiler(); if (metrics != null && prof != null) metrics.record(getName(pt.getParticipant()) + "-resume", prof.getPartialInMillis()); if (prof != null) prof.reenable(); pausedCounter.decrementAndGet(); } } else pt = null; if (pt == null) { int running = getActiveTransactions(); if (maxActiveSessions > 0 && running >= maxActiveSessions) { getLog().warn ( Thread.currentThread().getName() + ": emergency retry, running-sessions=" + running + ", max-active-sessions=" + maxActiveSessions ); psp.out (RETRY_QUEUE, obj, retryTimeout); checkRetryTask(); continue; } abort = false; id = nextId (); members = new ArrayList (); iter = getParticipants (DEFAULT_GROUP).iterator(); activeTransactions.incrementAndGet(); } transactionActive = true; if (debug) { if (evt == null) { evt = getLog().createLogEvent("debug", Thread.currentThread().getName() + ":" + Long.toString(id) + (pt != null ? " [resuming]" : "") ); if (debugContext) { evt.addMessage (context); } } if (prof == null) prof = new Profiler(); else prof.checkPoint("resume"); } snapshot (id, context, PREPARING); setThreadLocal(id, context); action = prepare (session, id, context, members, iter, abort, evt, prof); removeThreadLocal(); switch (action) { case PAUSE: paused = true; if (id % TIMER_PURGE_INTERVAL == 0) timer.purge(); pausedCounter.incrementAndGet(); break; case PREPARED: if (members.size() > 0) { setState(id, COMMITTING); setThreadLocal(id, context); commit(session, id, context, members, false, evt, prof); removeThreadLocal(); } break; case ABORTED: if (members.size() > 0) { setThreadLocal(id, context); abort(session, id, context, members, false, evt, prof); removeThreadLocal(); } break; case RETRY: psp.out (RETRY_QUEUE, context); checkRetryTask(); break; case NO_JOIN: break; } if ((action & PAUSE) == 0) { snapshot (id, null, DONE); if (id == tail) { checkTail (); } else { purge (id, false); } tps.tick(); } } catch (Throwable t) { if (evt == null) getLog().fatal (t); else evt.addMessage (t); } finally { removeThreadLocal(); if (transactionActive && !paused) activeTransactions.decrementAndGet(); if (hasStatusListeners) { notifyStatusListeners ( session, paused ? TransactionStatusEvent.State.PAUSED : TransactionStatusEvent.State.DONE, id, "", context); } if (evt != null && (action == PREPARED || action == ABORTED || (action == -1 && prof != null))) { switch (action) { case PREPARED : evt.setTag("commit"); break; case ABORTED : evt.setTag ("abort"); break; case -1: evt.setTag ("undefined"); break; } if (getInTransit() > Math.max(maxActiveSessions, activeSessions.get()) * 100) { evt.addMessage("WARNING: IN-TRANSIT TOO HIGH"); } evt.addMessage ( String.format (" %s, elapsed=%dms", tmInfo(), prof != null ? prof.getElapsedInMillis() : -1 ) ); if (prof != null) evt.addMessage (prof); try { Logger.log(freeze(context, evt, prof)); } catch (Throwable t) { getLog().error(t); } } } } threads.remove(thread); int currentActiveSessions = activeSessions.decrementAndGet(); getLog().info ("stop " + Thread.currentThread() + ", active sessions=" + currentActiveSessions); }  @Override void initService(); @Override void startService(); @Override void stopService(); void queue(Serializable context); void push(Serializable context); @SuppressWarnings("unused") String getQueueName(); Space getSpace(); Space getInputSpace(); Space getPersistentSpace(); @Override void run(); @Override long getTail(); @Override long getHead(); long getInTransit(); @Override void setConfiguration(Configuration cfg); void addListener(TransactionStatusListener l); void removeListener(TransactionStatusListener l); TPS getTPS(); @Override String getTPSAsString(); @Override float getTPSAvg(); @Override int getTPSPeak(); @Override Date getTPSPeakWhen(); @Override long getTPSElapsed(); @Override void resetTPS(); @Override Metrics getMetrics(); @Override void dump(PrintStream ps, String indent); TransactionParticipant createParticipant(Element e); @Override int getOutstandingTransactions(); @Override void setDebug(boolean debug); @Override boolean getDebugContext(); @Override void setDebugContext(boolean debugContext); @Override boolean getDebug(); @Override int getActiveSessions(); int getPausedCounter(); int getActiveTransactions(); int getMaxSessions(); static Serializable getSerializable(); static T getContext(); static Long getId(); static final String HEAD; static final String TAIL; static final String CONTEXT; static final String STATE; static final String GROUPS; static final String TAILLOCK; static final String RETRY_QUEUE; static final Integer PREPARING; static final Integer COMMITTING; static final Integer DONE; static final String DEFAULT_GROUP; static final long MAX_PARTICIPANTS; static final long MAX_WAIT; static final long TIMER_PURGE_INTERVAL; }