BigtableStorage implements Closeable { void writeEvent(SequenceEvent sequenceEvent) throws IOException { try { retryer.call(() -> { try (final Table eventsTable = connection.getTable(EVENTS_TABLE_NAME)) { final String workflowInstanceKey = sequenceEvent.event().workflowInstance().toKey(); final String keyString = String.format("%s#%08d", workflowInstanceKey, sequenceEvent.counter()); final byte[] key = Bytes.toBytes(keyString); final Put put = new Put(key, sequenceEvent.timestamp()); final byte[] eventBytes = serialize(sequenceEvent.event()).toByteArray(); put.addColumn(EVENT_CF, EVENT_QUALIFIER, eventBytes); eventsTable.put(put); return null; } }); } catch (ExecutionException | RetryException e) { var cause = e.getCause(); if (cause instanceof IOException) { throw (IOException) cause; } else { throw new RuntimeException(cause); } } } BigtableStorage(Connection connection); @VisibleForTesting  BigtableStorage(Connection connection,
                  ExecutorService executor,
                  WaitStrategy waitStrategy,
                  StopStrategy retryStopStrategy); @Override void close(); static final TableName EVENTS_TABLE_NAME; }
@Test public void shouldProduceIOExceptionIfTooManyPutRetries() throws Exception { setUp(MAX_BIGTABLE_RETRIES); thrown.expect(IOException.class); thrown.expectMessage(containsString("Something went wrong in performing put operation")); storage.writeEvent(SequenceEvent.create(Event.success(WFI1), 1, 0)); }
@Test public void shouldNotProduceIOExceptionIfPutRetrySucceeds() throws Exception { setUp(MAX_BIGTABLE_RETRIES - 1); storage.writeEvent(SequenceEvent.create(Event.success(WFI1), 1, 0)); }
@Test public void shouldProduceRuntimeExceptionWrappingRuntimeException() throws Exception { setUp(0); var e = new RuntimeException(); doThrow(e).when(bigtable).getTable(any()); thrown.expect(RuntimeException.class); thrown.expectCause(is(e)); storage.writeEvent(SequenceEvent.create(Event.success(WFI1), 1, 0)); }