JobNodeStorage { public void executeInTransaction(final TransactionExecutionCallback callback) { try { List<CuratorOp> operations = new LinkedList<>(); CuratorFramework client = getClient(); TransactionOp transactionOp = client.transactionOp(); operations.add(transactionOp.check().forPath("/")); operations.addAll(callback.createCuratorOperators(transactionOp)); client.transaction().forOperations(operations); } catch (final Exception ex) { RegExceptionHandler.handleException(ex); } } JobNodeStorage(final CoordinatorRegistryCenter regCenter, final String jobName); boolean isJobNodeExisted(final String node); boolean isJobRootNodeExisted(); String getJobNodeData(final String node); String getJobNodeDataDirectly(final String node); List<String> getJobNodeChildrenKeys(final String node); String getJobRootNodeData(); void createJobNodeIfNeeded(final String node); void removeJobNodeIfExisted(final String node); void fillJobNode(final String node, final Object value); void fillEphemeralJobNode(final String node, final Object value); void updateJobNode(final String node, final Object value); void replaceJobNode(final String node, final Object value); void replaceJobRootNode(final Object value); void executeInTransaction(final TransactionExecutionCallback callback); void executeInLeader(final String latchNode, final LeaderExecutionCallback callback); void addConnectionStateListener(final ConnectionStateListener listener); void addDataListener(final CuratorCacheListener listener); long getRegistryCenterTime();  }
@Test public void assertExecuteInTransactionSuccess() throws Exception { CuratorFramework client = mock(CuratorFramework.class); when(regCenter.getRawClient()).thenReturn(client); TransactionOp transactionOp = mockTransactionOp(client); CuratorOp checkOp = mockCheckOp(transactionOp); CuratorOp createOp = mockCreateOp(transactionOp); CuratorMultiTransaction transaction = mockTransaction(client); jobNodeStorage.executeInTransaction(input -> Collections.singletonList(input.create().forPath("/test_transaction"))); verify(transaction).forOperations(Arrays.asList(checkOp, createOp)); }
@Test(expected = RuntimeException.class) public void assertExecuteInTransactionFailure() throws Exception { CuratorFramework client = mock(CuratorFramework.class); when(regCenter.getRawClient()).thenReturn(client); TransactionOp transactionOp = mockTransactionOp(client); CuratorOp checkOp = mockCheckOp(transactionOp); CuratorOp createFailedOp = mockCreateFailedOp(transactionOp); CuratorMultiTransaction transaction = mockTransaction(client); try { jobNodeStorage.executeInTransaction(input -> Collections.singletonList(input.create().forPath("/test_transaction"))); } finally { verify(transaction, times(0)).forOperations(Arrays.asList(checkOp, createFailedOp)); } }