@Test public void testSuspendFeed() throws Exception { final int numTransactions = WaltzClientConfig.DEFAULT_MAX_CONCURRENT_TRANSACTIONS; Map<Integer, MockServerPartition> serverPartitions = Collections.singletonMap(0, new MockServerPartition(0)); MockDriver driver1 = new MockDriver(1, serverPartitions); DummyCallbacks callbacks1 = new DummyCallbacks(); callbacks1.setClientHighWaterMark(0, -1L); driver1.initialize(callbacks1, config); MockDriver driver2 = new MockDriver(2, serverPartitions); DummyCallbacks callbacks2 = new DummyCallbacks(); callbacks2.setClientHighWaterMark(0, -1L); driver2.initialize(callbacks2, config); driver2.suspendFeed(); StreamClient streamClient1 = driver1.getStreamClient(); StreamClient streamClient2 = driver2.getStreamClient(); ArrayList<TransactionFuture> futures = new ArrayList<>(); int numSuccess = 0; for (int i = 0; i < numTransactions; i++) { DummyContext context = new DummyContext(0, "transaction" + i); TransactionBuilderImpl builder = streamClient1.getTransactionBuilder(context); context.execute(builder); AppendRequest transaction = builder.buildRequest(); futures.add(streamClient1.append(transaction, context)); } streamClient1.flushTransactions(); callbacks1.awaitHighWaterMark(0, (long) numTransactions - 1L, 1000); for (TransactionFuture future: futures) { if (future.get(1000, TimeUnit.MILLISECONDS)) { numSuccess++; } } assertEquals(numTransactions, numSuccess); assertEquals((long) numTransactions - 1L, callbacks1.getClientHighWaterMark(0)); Uninterruptibly.sleep(50); assertEquals(-1L, callbacks2.getClientHighWaterMark(0)); driver2.resumeFeed(); DummyContext context = new DummyContext(0, "extra transaction"); TransactionBuilderImpl builder = streamClient2.getTransactionBuilder(context); context.execute(builder); AppendRequest request = builder.buildRequest(); streamClient2.append(request, context); streamClient2.flushTransactions(); callbacks2.awaitHighWaterMark(0, (long) numTransactions, 1000); assertEquals((long) numTransactions, callbacks2.getClientHighWaterMark(0)); }
public void suspendFeed() { streamClient.suspendFeed(); }
MockDriver implements WaltzClientDriver { public void suspendFeed() { streamClient.suspendFeed(); } }
MockDriver implements WaltzClientDriver { public void suspendFeed() { streamClient.suspendFeed(); } MockDriver();  MockDriver(int clientId, Map<Integer, MockServerPartition> serverPartitions); }
MockDriver implements WaltzClientDriver { public void suspendFeed() { streamClient.suspendFeed(); } MockDriver();  MockDriver(int clientId, Map<Integer, MockServerPartition> serverPartitions); void initialize(WaltzClientCallbacks callbacks, WaltzClientConfig config); @Override ClusterManager getClusterManager(); @Override ManagedClient getManagedClient(); @Override RpcClient getRpcClient(); @Override StreamClient getStreamClient(); @Override void close(); void setMaxDelay(long maxDelayMillis); void setFaultRate(double faultRate); void forceNextAppendFail(); void forceNextLockFail(); void suspendFeed(); void resumeFeed(); }
MockDriver implements WaltzClientDriver { public void suspendFeed() { streamClient.suspendFeed(); } MockDriver();  MockDriver(int clientId, Map<Integer, MockServerPartition> serverPartitions); void initialize(WaltzClientCallbacks callbacks, WaltzClientConfig config); @Override ClusterManager getClusterManager(); @Override ManagedClient getManagedClient(); @Override RpcClient getRpcClient(); @Override StreamClient getStreamClient(); @Override void close(); void setMaxDelay(long maxDelayMillis); void setFaultRate(double faultRate); void forceNextAppendFail(); void forceNextLockFail(); void suspendFeed(); void resumeFeed();  }