ConcurrentRowCallables { public void addCallable(RowCallable<R> callable) throws IOException { TableAndRow tableAndRow = callable.getTableAndRow(); if (this.futureMaps.containsKey(tableAndRow) || this.exceptionMaps.containsKey(tableAndRow)) { throw new ThemisFatalException("add duplicated row callable, tableAndRow=" + tableAndRow); } try { Future<R> future = this.threadPool.submit(callable); this.futureMaps.put(tableAndRow, future); } catch (Throwable e) { exceptionMaps.put(tableAndRow, new IOException(e)); } } ConcurrentRowCallables(ExecutorService threadPool); void addCallable(RowCallable<R> callable); void waitForResult(); Map<TableAndRow, R> getResults(); R getResult(byte[] tableName, byte[] rowkey); Map<TableAndRow, IOException> getExceptions(); IOException getException(byte[] tableName, byte[] rowkey);  }
@Test public void testDuplicateRowCallable() throws IOException { ConcurrentRowCallables<byte[]> callables = new ConcurrentRowCallables<byte[]>(threadPool); callables.addCallable(new TimeoutRowCallable(TABLENAME, ROW)); callables.addCallable(new TimeoutRowCallable(TABLENAME, ANOTHER_ROW)); callables.addCallable(new TimeoutRowCallable(TABLENAME, ANOTHER_FAMILY)); callables.addCallable(new TimeoutRowCallable(TABLENAME, ANOTHER_QUALIFIER)); try { callables.addCallable(new TimeoutRowCallable(TABLENAME, ROW)); Assert.fail(); } catch (ThemisFatalException e) {} try { callables.addCallable(new TimeoutRowCallable(TABLENAME, ANOTHER_QUALIFIER)); Assert.fail(); } catch (ThemisFatalException e) {} }