@Test public void test() throws Exception { final int numTransactions = 10; List<String> expected = new ArrayList<>(); InternalRpcClient internalRpcClient = getInternalRpcClient(WaltzClientConfig.DEFAULT_MAX_CONCURRENT_TRANSACTIONS); InternalStreamClient internalStreamClient = getInternalStreamClient(true, WaltzClientConfig.DEFAULT_MAX_CONCURRENT_TRANSACTIONS, internalRpcClient); assertEquals(allPartitions, internalRpcClient.getActivePartitions()); for (int i = 0; i < numTransactions; i++) { String data = "transaction" + i; expected.add(data); MockContext context = MockContext.builder().header(0).data(data).build(); TransactionBuilderImpl transactionBuilder = internalStreamClient.getTransactionBuilder(context); context.execute(transactionBuilder); TransactionFuture future = internalStreamClient.append(transactionBuilder.buildRequest(), context); assertTrue(future.get()); } ArrayList<Integer> ids = new ArrayList<>(); for (int i = 0; i < numTransactions; i++) { ids.add(i); } for (Integer i : ids) { byte[] data = Uninterruptibly.call(() -> internalRpcClient.getTransactionData(0, i).get()); assertEquals(expected.get(i), new String(data, UTF_8)); } Collections.shuffle(ids); for (Integer i : ids) { byte[] data = Uninterruptibly.call(() -> internalRpcClient.getTransactionData(0, i).get()); assertEquals(expected.get(i), new String(data, UTF_8)); } }
@Override public Future<byte[]> getTransactionData(int partitionId, long transactionId) { return getPartition(partitionId).getTransactionData(transactionId); }
InternalRpcClient extends InternalBaseClient implements RpcClient { @Override public Future<byte[]> getTransactionData(int partitionId, long transactionId) { return getPartition(partitionId).getTransactionData(transactionId); } }
InternalRpcClient extends InternalBaseClient implements RpcClient { @Override public Future<byte[]> getTransactionData(int partitionId, long transactionId) { return getPartition(partitionId).getTransactionData(transactionId); } InternalRpcClient(SslContext sslCtx, int maxConcurrentTransactions, WaltzClientCallbacks callbacks); }
InternalRpcClient extends InternalBaseClient implements RpcClient { @Override public Future<byte[]> getTransactionData(int partitionId, long transactionId) { return getPartition(partitionId).getTransactionData(transactionId); } InternalRpcClient(SslContext sslCtx, int maxConcurrentTransactions, WaltzClientCallbacks callbacks); @Override void onMountingPartition(final WaltzNetworkClient networkClient, final Partition partition); @Override Future<byte[]> getTransactionData(int partitionId, long transactionId); @Override Future<Long> getHighWaterMark(int partitionId); @Override Future<Map<Endpoint, Map<String, Boolean>>> checkServerConnections(Set<Endpoint> serverEndpoints); @Override Future<List<Integer>> getServerPartitionAssignments(Endpoint serverEndpoint); @Override Future<Boolean> addPreferredPartition(Endpoint serverEndpoint, int partitionId); @Override Future<Boolean> removePreferredPartition(Endpoint serverEndpoint,
                                                    int partitionId); }
InternalRpcClient extends InternalBaseClient implements RpcClient { @Override public Future<byte[]> getTransactionData(int partitionId, long transactionId) { return getPartition(partitionId).getTransactionData(transactionId); } InternalRpcClient(SslContext sslCtx, int maxConcurrentTransactions, WaltzClientCallbacks callbacks); @Override void onMountingPartition(final WaltzNetworkClient networkClient, final Partition partition); @Override Future<byte[]> getTransactionData(int partitionId, long transactionId); @Override Future<Long> getHighWaterMark(int partitionId); @Override Future<Map<Endpoint, Map<String, Boolean>>> checkServerConnections(Set<Endpoint> serverEndpoints); @Override Future<List<Integer>> getServerPartitionAssignments(Endpoint serverEndpoint); @Override Future<Boolean> addPreferredPartition(Endpoint serverEndpoint, int partitionId); @Override Future<Boolean> removePreferredPartition(Endpoint serverEndpoint,
                                                    int partitionId);  }