RemoteTable extends BaseReadWriteTable<K, V> implements ReadWriteTable<K, V>, AsyncReadWriteTable<K, V> { @Override public CompletableFuture<V> getAsync(K key, Object ... args) { Preconditions.checkNotNull(key, "null key"); return instrument(() -> asyncTable.getAsync(key, args), metrics.numGets, metrics.getNs) .handle((result, e) -> { if (e != null) { throw new SamzaException("Failed to get the records for " + key, e); } if (result == null) { incCounter(metrics.numMissedLookups); } return result; }); } RemoteTable(
      String tableId,
      TableReadFunction readFn,
      TableWriteFunction writeFn,
      TableRateLimiter<K, V> readRateLimiter,
      TableRateLimiter<K, V> writeRateLimiter,
      ExecutorService rateLimitingExecutor,
      TableRetryPolicy readRetryPolicy,
      TableRetryPolicy writeRetryPolicy,
      ScheduledExecutorService retryExecutor,
      BatchProvider<K, V> batchProvider,
      ScheduledExecutorService batchExecutor,
      ExecutorService callbackExecutor); @Override V get(K key, Object ... args); @Override CompletableFuture<V> getAsync(K key, Object ... args); @Override Map<K, V> getAll(List<K> keys, Object ... args); @Override CompletableFuture<Map<K, V>> getAllAsync(List<K> keys, Object ... args); @Override T read(int opId, Object ... args); @Override CompletableFuture<T> readAsync(int opId, Object ... args); @Override void put(K key, V value, Object ... args); @Override CompletableFuture<Void> putAsync(K key, V value, Object ... args); @Override void putAll(List<Entry<K, V>> entries, Object ... args); @Override CompletableFuture<Void> putAllAsync(List<Entry<K, V>> records, Object ... args); @Override void delete(K key, Object ... args); @Override CompletableFuture<Void> deleteAsync(K key, Object ... args); @Override void deleteAll(List<K> keys, Object ... args); @Override CompletableFuture<Void> deleteAllAsync(List<K> keys, Object ... args); @Override T write(int opId, Object ... args); @Override CompletableFuture<T> writeAsync(int opId, Object... args); @Override void init(Context context); @Override void flush(); @Override void close(); TableReadFunction<K, V> getReadFunction(); TableWriteFunction<K, V> getWriteFunction();  }
@Test public void testGetAsync() { doTestGet(false, false, false); }
@Test public void testGetMultipleTables() { TableReadFunction<String, String> readFn1 = mock(TableReadFunction.class); TableReadFunction<String, String> readFn2 = mock(TableReadFunction.class); doReturn(CompletableFuture.completedFuture("bar1")).when(readFn1).getAsync(anyString()); doReturn(CompletableFuture.completedFuture("bar2")).when(readFn1).getAsync(anyString()); RemoteTable<String, String> table1 = getTable("testGetMultipleTables-1", readFn1, null, false); RemoteTable<String, String> table2 = getTable("testGetMultipleTables-2", readFn2, null, false); CompletableFuture<String> future1 = table1.getAsync("foo1"); CompletableFuture<String> future2 = table2.getAsync("foo2"); CompletableFuture.allOf(future1, future2) .thenAccept(u -> { Assert.assertEquals(future1.join(), "bar1"); Assert.assertEquals(future2.join(), "bar1"); }); }
@Test public void testGetWithCallbackExecutor() { TableReadFunction<String, String> readFn = mock(TableReadFunction.class); doReturn(CompletableFuture.completedFuture("bar")).when(readFn).getAsync(anyString()); RemoteTable<String, String> table = getTable("testGetWithCallbackExecutor", readFn, null, Executors.newSingleThreadExecutor(), false); Thread testThread = Thread.currentThread(); table.getAsync("foo").thenAccept(result -> { Assert.assertEquals("bar", result); Assert.assertNotSame(testThread, Thread.currentThread()); }); }