@Test public void testRegister() throws Exception { monitor.start(); ReqId reqId1 = new ReqId(0, 1); ReqId reqId2 = new ReqId(0, 2); ReqId reqId3 = new ReqId(0, 3); ReqId reqId4 = new ReqId(0, 4); TransactionFuture future1 = monitor.register(reqId1, context, timeout); assertFalse(future1.isDone()); TransactionFuture futureDup = monitor.register(reqId1, context, timeout); assertTrue(futureDup.isDone()); assertFalse(futureDup.get()); TransactionFuture future2 = monitor.register(reqId2, context, timeout); assertFalse(future2.isDone()); TransactionFuture future3 = monitor.register(reqId3, context, timeout); assertFalse(future3.isDone()); TransactionFuture future4 = monitor.register(reqId4, context, timeout); assertFalse(future3.isDone()); monitor.committed(reqId1); assertTrue(future1.isDone()); assertFalse(future2.isDone()); assertFalse(future3.isDone()); assertFalse(future4.isDone()); monitor.committed(reqId4); assertTrue(future2.isDone()); assertTrue(future3.isDone()); assertTrue(future4.isDone()); assertFalse(future2.get()); assertFalse(future3.get()); assertTrue(future4.get()); }
public TransactionFuture register(ReqId reqId, TransactionContext context, long registrationTimeout) { synchronized (this) { final long due = System.currentTimeMillis() + registrationTimeout; while (state == State.STARTED && (numRegistered >= maxConcurrentTransactions)) { long remaining = due - System.currentTimeMillis(); if (remaining > 0) { try { wait(remaining); } catch (InterruptedException ex) { Thread.interrupted(); } } else { logger.warn("transaction registration timeout: partitionId=" + reqId.partitionId() + " elapsed=" + registrationTimeout + "ms"); return null; } } TransactionFuture future = new TransactionFuture(reqId, context); if (state == State.STARTED) { if (!registered.containsKey(reqId)) { lastReqId = reqId; lastTimestamp = System.currentTimeMillis(); registered.put(reqId, future); numRegistered++; if (numRegistered >= maxConcurrentTransactions) { logger.debug("transaction monitor reached capacity"); } } else { logger.error("duplicate reqId: reqId=" + reqId); future.complete(false); } } else { if (state == State.STOPPED) { future.completeExceptionally(new PartitionInactiveException(reqId.partitionId())); } else { future.completeExceptionally(new ClientClosedException()); } } return future; } }
TransactionMonitor { public TransactionFuture register(ReqId reqId, TransactionContext context, long registrationTimeout) { synchronized (this) { final long due = System.currentTimeMillis() + registrationTimeout; while (state == State.STARTED && (numRegistered >= maxConcurrentTransactions)) { long remaining = due - System.currentTimeMillis(); if (remaining > 0) { try { wait(remaining); } catch (InterruptedException ex) { Thread.interrupted(); } } else { logger.warn("transaction registration timeout: partitionId=" + reqId.partitionId() + " elapsed=" + registrationTimeout + "ms"); return null; } } TransactionFuture future = new TransactionFuture(reqId, context); if (state == State.STARTED) { if (!registered.containsKey(reqId)) { lastReqId = reqId; lastTimestamp = System.currentTimeMillis(); registered.put(reqId, future); numRegistered++; if (numRegistered >= maxConcurrentTransactions) { logger.debug("transaction monitor reached capacity"); } } else { logger.error("duplicate reqId: reqId=" + reqId); future.complete(false); } } else { if (state == State.STOPPED) { future.completeExceptionally(new PartitionInactiveException(reqId.partitionId())); } else { future.completeExceptionally(new ClientClosedException()); } } return future; } } }
TransactionMonitor { public TransactionFuture register(ReqId reqId, TransactionContext context, long registrationTimeout) { synchronized (this) { final long due = System.currentTimeMillis() + registrationTimeout; while (state == State.STARTED && (numRegistered >= maxConcurrentTransactions)) { long remaining = due - System.currentTimeMillis(); if (remaining > 0) { try { wait(remaining); } catch (InterruptedException ex) { Thread.interrupted(); } } else { logger.warn("transaction registration timeout: partitionId=" + reqId.partitionId() + " elapsed=" + registrationTimeout + "ms"); return null; } } TransactionFuture future = new TransactionFuture(reqId, context); if (state == State.STARTED) { if (!registered.containsKey(reqId)) { lastReqId = reqId; lastTimestamp = System.currentTimeMillis(); registered.put(reqId, future); numRegistered++; if (numRegistered >= maxConcurrentTransactions) { logger.debug("transaction monitor reached capacity"); } } else { logger.error("duplicate reqId: reqId=" + reqId); future.complete(false); } } else { if (state == State.STOPPED) { future.completeExceptionally(new PartitionInactiveException(reqId.partitionId())); } else { future.completeExceptionally(new ClientClosedException()); } } return future; } } TransactionMonitor(int maxConcurrentTransactions); }
TransactionMonitor { public TransactionFuture register(ReqId reqId, TransactionContext context, long registrationTimeout) { synchronized (this) { final long due = System.currentTimeMillis() + registrationTimeout; while (state == State.STARTED && (numRegistered >= maxConcurrentTransactions)) { long remaining = due - System.currentTimeMillis(); if (remaining > 0) { try { wait(remaining); } catch (InterruptedException ex) { Thread.interrupted(); } } else { logger.warn("transaction registration timeout: partitionId=" + reqId.partitionId() + " elapsed=" + registrationTimeout + "ms"); return null; } } TransactionFuture future = new TransactionFuture(reqId, context); if (state == State.STARTED) { if (!registered.containsKey(reqId)) { lastReqId = reqId; lastTimestamp = System.currentTimeMillis(); registered.put(reqId, future); numRegistered++; if (numRegistered >= maxConcurrentTransactions) { logger.debug("transaction monitor reached capacity"); } } else { logger.error("duplicate reqId: reqId=" + reqId); future.complete(false); } } else { if (state == State.STOPPED) { future.completeExceptionally(new PartitionInactiveException(reqId.partitionId())); } else { future.completeExceptionally(new ClientClosedException()); } } return future; } } TransactionMonitor(int maxConcurrentTransactions); void close(); boolean start(); boolean stop(); boolean isStopped(); void clear(); int maxCapacity(); TransactionFuture register(ReqId reqId, TransactionContext context, long registrationTimeout); TransactionContext getTransactionContext(ReqId reqId); TransactionContext committed(ReqId reqId); void abort(ReqId reqId); void flush(ReqId reqId); TransactionFuture lastEnqueued(); long lastEnqueuedTime(); int registeredCount(); }
TransactionMonitor { public TransactionFuture register(ReqId reqId, TransactionContext context, long registrationTimeout) { synchronized (this) { final long due = System.currentTimeMillis() + registrationTimeout; while (state == State.STARTED && (numRegistered >= maxConcurrentTransactions)) { long remaining = due - System.currentTimeMillis(); if (remaining > 0) { try { wait(remaining); } catch (InterruptedException ex) { Thread.interrupted(); } } else { logger.warn("transaction registration timeout: partitionId=" + reqId.partitionId() + " elapsed=" + registrationTimeout + "ms"); return null; } } TransactionFuture future = new TransactionFuture(reqId, context); if (state == State.STARTED) { if (!registered.containsKey(reqId)) { lastReqId = reqId; lastTimestamp = System.currentTimeMillis(); registered.put(reqId, future); numRegistered++; if (numRegistered >= maxConcurrentTransactions) { logger.debug("transaction monitor reached capacity"); } } else { logger.error("duplicate reqId: reqId=" + reqId); future.complete(false); } } else { if (state == State.STOPPED) { future.completeExceptionally(new PartitionInactiveException(reqId.partitionId())); } else { future.completeExceptionally(new ClientClosedException()); } } return future; } } TransactionMonitor(int maxConcurrentTransactions); void close(); boolean start(); boolean stop(); boolean isStopped(); void clear(); int maxCapacity(); TransactionFuture register(ReqId reqId, TransactionContext context, long registrationTimeout); TransactionContext getTransactionContext(ReqId reqId); TransactionContext committed(ReqId reqId); void abort(ReqId reqId); void flush(ReqId reqId); TransactionFuture lastEnqueued(); long lastEnqueuedTime(); int registeredCount();  }