@Test public void repeatedProcessBatchOnlyMakesOneCallWithDuplicatesIfSuccessful() { KeyAlreadyExistsException exception = new KeyAlreadyExistsException("boo", ImmutableList.of(ENCODING_STRATEGY.encodeStartTimestampAsCell(5L))); doNothing() .doThrow(exception) .when(mockTransactionService) .putUnlessExistsMultiple(anyMap()); int numRequests = 100; List<BatchElement<WriteBatchingTransactionService.TimestampPair, Void>> batchedRequest = IntStream.range(0, numRequests) .mapToObj(unused -> TestTransactionBatchElement.of(5L, 9L)) .collect(Collectors.toList()); WriteBatchingTransactionService.processBatch(mockTransactionService, batchedRequest); AtomicInteger successCount = new AtomicInteger(); AtomicInteger failureCount = new AtomicInteger(); getResultsTrackingOutcomes(batchedRequest, successCount, failureCount); verify(mockTransactionService, times(1)).putUnlessExistsMultiple(anyMap()); verify(mockTransactionService, atLeastOnce()).getEncodingStrategy(); assertThat(successCount).hasValue(1); assertThat(failureCount).hasValue(numRequests - 1); }
@VisibleForTesting static void processBatch( EncodingTransactionService delegate, List<BatchElement<TimestampPair, Void>> batchElements) { Multimap<Long, BatchElement<TimestampPair, Void>> startTimestampKeyedBatchElements = MultimapBuilder.hashKeys().hashSetValues().build(); batchElements.forEach(batchElement -> startTimestampKeyedBatchElements.put( batchElement.argument().startTimestamp(), batchElement)); while (!startTimestampKeyedBatchElements.isEmpty()) { Map<Long, BatchElement<TimestampPair, Void>> batch = extractSingleBatchForQuerying(startTimestampKeyedBatchElements); try { delegate.putUnlessExistsMultiple(KeyedStream.stream(batch) .map(batchElement -> batchElement.argument().commitTimestamp()) .collectToMap()); markBatchSuccessful(startTimestampKeyedBatchElements, batch); markAllRemainingRequestsAsFailed(delegate, startTimestampKeyedBatchElements.values()); return; } catch (KeyAlreadyExistsException exception) { handleFailedTimestamps(delegate, startTimestampKeyedBatchElements, batch, exception); handleSuccessfulTimestamps(delegate, startTimestampKeyedBatchElements, batch, exception); } } }
WriteBatchingTransactionService implements TransactionService { @VisibleForTesting static void processBatch( EncodingTransactionService delegate, List<BatchElement<TimestampPair, Void>> batchElements) { Multimap<Long, BatchElement<TimestampPair, Void>> startTimestampKeyedBatchElements = MultimapBuilder.hashKeys().hashSetValues().build(); batchElements.forEach(batchElement -> startTimestampKeyedBatchElements.put( batchElement.argument().startTimestamp(), batchElement)); while (!startTimestampKeyedBatchElements.isEmpty()) { Map<Long, BatchElement<TimestampPair, Void>> batch = extractSingleBatchForQuerying(startTimestampKeyedBatchElements); try { delegate.putUnlessExistsMultiple(KeyedStream.stream(batch) .map(batchElement -> batchElement.argument().commitTimestamp()) .collectToMap()); markBatchSuccessful(startTimestampKeyedBatchElements, batch); markAllRemainingRequestsAsFailed(delegate, startTimestampKeyedBatchElements.values()); return; } catch (KeyAlreadyExistsException exception) { handleFailedTimestamps(delegate, startTimestampKeyedBatchElements, batch, exception); handleSuccessfulTimestamps(delegate, startTimestampKeyedBatchElements, batch, exception); } } } }
WriteBatchingTransactionService implements TransactionService { @VisibleForTesting static void processBatch( EncodingTransactionService delegate, List<BatchElement<TimestampPair, Void>> batchElements) { Multimap<Long, BatchElement<TimestampPair, Void>> startTimestampKeyedBatchElements = MultimapBuilder.hashKeys().hashSetValues().build(); batchElements.forEach(batchElement -> startTimestampKeyedBatchElements.put( batchElement.argument().startTimestamp(), batchElement)); while (!startTimestampKeyedBatchElements.isEmpty()) { Map<Long, BatchElement<TimestampPair, Void>> batch = extractSingleBatchForQuerying(startTimestampKeyedBatchElements); try { delegate.putUnlessExistsMultiple(KeyedStream.stream(batch) .map(batchElement -> batchElement.argument().commitTimestamp()) .collectToMap()); markBatchSuccessful(startTimestampKeyedBatchElements, batch); markAllRemainingRequestsAsFailed(delegate, startTimestampKeyedBatchElements.values()); return; } catch (KeyAlreadyExistsException exception) { handleFailedTimestamps(delegate, startTimestampKeyedBatchElements, batch, exception); handleSuccessfulTimestamps(delegate, startTimestampKeyedBatchElements, batch, exception); } } } private  WriteBatchingTransactionService(
            EncodingTransactionService delegate, DisruptorAutobatcher<TimestampPair, Void> autobatcher); }
WriteBatchingTransactionService implements TransactionService { @VisibleForTesting static void processBatch( EncodingTransactionService delegate, List<BatchElement<TimestampPair, Void>> batchElements) { Multimap<Long, BatchElement<TimestampPair, Void>> startTimestampKeyedBatchElements = MultimapBuilder.hashKeys().hashSetValues().build(); batchElements.forEach(batchElement -> startTimestampKeyedBatchElements.put( batchElement.argument().startTimestamp(), batchElement)); while (!startTimestampKeyedBatchElements.isEmpty()) { Map<Long, BatchElement<TimestampPair, Void>> batch = extractSingleBatchForQuerying(startTimestampKeyedBatchElements); try { delegate.putUnlessExistsMultiple(KeyedStream.stream(batch) .map(batchElement -> batchElement.argument().commitTimestamp()) .collectToMap()); markBatchSuccessful(startTimestampKeyedBatchElements, batch); markAllRemainingRequestsAsFailed(delegate, startTimestampKeyedBatchElements.values()); return; } catch (KeyAlreadyExistsException exception) { handleFailedTimestamps(delegate, startTimestampKeyedBatchElements, batch, exception); handleSuccessfulTimestamps(delegate, startTimestampKeyedBatchElements, batch, exception); } } } private  WriteBatchingTransactionService(
            EncodingTransactionService delegate, DisruptorAutobatcher<TimestampPair, Void> autobatcher); static TransactionService create(EncodingTransactionService delegate); @CheckForNull @Override Long get(long startTimestamp); @Override Map<Long, Long> get(Iterable<Long> startTimestamps); @Override ListenableFuture<Long> getAsync(long startTimestamp); @Override ListenableFuture<Map<Long, Long>> getAsync(Iterable<Long> startTimestamps); @Override void putUnlessExists(long startTimestamp, long commitTimestamp); @Override void close(); }
WriteBatchingTransactionService implements TransactionService { @VisibleForTesting static void processBatch( EncodingTransactionService delegate, List<BatchElement<TimestampPair, Void>> batchElements) { Multimap<Long, BatchElement<TimestampPair, Void>> startTimestampKeyedBatchElements = MultimapBuilder.hashKeys().hashSetValues().build(); batchElements.forEach(batchElement -> startTimestampKeyedBatchElements.put( batchElement.argument().startTimestamp(), batchElement)); while (!startTimestampKeyedBatchElements.isEmpty()) { Map<Long, BatchElement<TimestampPair, Void>> batch = extractSingleBatchForQuerying(startTimestampKeyedBatchElements); try { delegate.putUnlessExistsMultiple(KeyedStream.stream(batch) .map(batchElement -> batchElement.argument().commitTimestamp()) .collectToMap()); markBatchSuccessful(startTimestampKeyedBatchElements, batch); markAllRemainingRequestsAsFailed(delegate, startTimestampKeyedBatchElements.values()); return; } catch (KeyAlreadyExistsException exception) { handleFailedTimestamps(delegate, startTimestampKeyedBatchElements, batch, exception); handleSuccessfulTimestamps(delegate, startTimestampKeyedBatchElements, batch, exception); } } } private  WriteBatchingTransactionService(
            EncodingTransactionService delegate, DisruptorAutobatcher<TimestampPair, Void> autobatcher); static TransactionService create(EncodingTransactionService delegate); @CheckForNull @Override Long get(long startTimestamp); @Override Map<Long, Long> get(Iterable<Long> startTimestamps); @Override ListenableFuture<Long> getAsync(long startTimestamp); @Override ListenableFuture<Map<Long, Long>> getAsync(Iterable<Long> startTimestamps); @Override void putUnlessExists(long startTimestamp, long commitTimestamp); @Override void close();  }