@Test public void throwsWhenCheckAndSetFailsWithNoDetailsAndNoValueInKvs() { KeyValueService mockKvs = mock(KeyValueService.class); KeyValueServiceCoordinationStore<String> store = coordinationStoreForKvs(mockKvs); doThrow(new CheckAndSetException("No detail")).when(mockKvs).checkAndSet(any(CheckAndSetRequest.class)); when(mockKvs.get(eq(AtlasDbConstants.COORDINATION_TABLE), anyMap())).thenReturn(ImmutableMap.of()); assertThatThrownBy(() -> store.checkAndSetCoordinationValue(Optional.empty(), SEQUENCE_AND_BOUND_2)) .as("Failed but no value in KVS") .isInstanceOf(IllegalStateException.class) .hasMessageContaining("Failed to check and set coordination value"); }
@SuppressWarnings("OptionalUsedAsFieldOrParameterType") @VisibleForTesting CheckAndSetResult<SequenceAndBound> checkAndSetCoordinationValue( Optional<SequenceAndBound> oldValue, SequenceAndBound newValue) { if (oldValue.map(presentOldValue -> Objects.equals(presentOldValue, newValue)).orElse(false)) { return ImmutableCheckAndSetResult.of(true, ImmutableList.of(newValue)); } CheckAndSetRequest request = new CheckAndSetRequest.Builder() .table(AtlasDbConstants.COORDINATION_TABLE) .cell(getCoordinationValueCell()) .oldValue(oldValue.map(this::serializeSequenceAndBound)) .newValue(serializeSequenceAndBound(newValue)) .build(); try { kvs.checkAndSet(request); return ImmutableCheckAndSetResult.of(true, ImmutableList.of(newValue)); } catch (CheckAndSetException e) { if (e.getActualValues() != null) { return ImmutableCheckAndSetResult.of(false, e.getActualValues() .stream() .map(this::deserializeSequenceAndBound) .collect(Collectors.toList())); } return getFailedResultWithMostRecentValueAndBound(oldValue, newValue); } }
KeyValueServiceCoordinationStore implements CoordinationStore<T> { @SuppressWarnings("OptionalUsedAsFieldOrParameterType") @VisibleForTesting CheckAndSetResult<SequenceAndBound> checkAndSetCoordinationValue( Optional<SequenceAndBound> oldValue, SequenceAndBound newValue) { if (oldValue.map(presentOldValue -> Objects.equals(presentOldValue, newValue)).orElse(false)) { return ImmutableCheckAndSetResult.of(true, ImmutableList.of(newValue)); } CheckAndSetRequest request = new CheckAndSetRequest.Builder() .table(AtlasDbConstants.COORDINATION_TABLE) .cell(getCoordinationValueCell()) .oldValue(oldValue.map(this::serializeSequenceAndBound)) .newValue(serializeSequenceAndBound(newValue)) .build(); try { kvs.checkAndSet(request); return ImmutableCheckAndSetResult.of(true, ImmutableList.of(newValue)); } catch (CheckAndSetException e) { if (e.getActualValues() != null) { return ImmutableCheckAndSetResult.of(false, e.getActualValues() .stream() .map(this::deserializeSequenceAndBound) .collect(Collectors.toList())); } return getFailedResultWithMostRecentValueAndBound(oldValue, newValue); } } }
KeyValueServiceCoordinationStore implements CoordinationStore<T> { @SuppressWarnings("OptionalUsedAsFieldOrParameterType") @VisibleForTesting CheckAndSetResult<SequenceAndBound> checkAndSetCoordinationValue( Optional<SequenceAndBound> oldValue, SequenceAndBound newValue) { if (oldValue.map(presentOldValue -> Objects.equals(presentOldValue, newValue)).orElse(false)) { return ImmutableCheckAndSetResult.of(true, ImmutableList.of(newValue)); } CheckAndSetRequest request = new CheckAndSetRequest.Builder() .table(AtlasDbConstants.COORDINATION_TABLE) .cell(getCoordinationValueCell()) .oldValue(oldValue.map(this::serializeSequenceAndBound)) .newValue(serializeSequenceAndBound(newValue)) .build(); try { kvs.checkAndSet(request); return ImmutableCheckAndSetResult.of(true, ImmutableList.of(newValue)); } catch (CheckAndSetException e) { if (e.getActualValues() != null) { return ImmutableCheckAndSetResult.of(false, e.getActualValues() .stream() .map(this::deserializeSequenceAndBound) .collect(Collectors.toList())); } return getFailedResultWithMostRecentValueAndBound(oldValue, newValue); } } @VisibleForTesting  KeyValueServiceCoordinationStore(
            ObjectMapper objectMapper,
            KeyValueService kvs,
            byte[] coordinationRow,
            LongSupplier sequenceNumberSupplier,
            BiPredicate<T, T> shouldReuseExtantValue,
            Class<T> clazz); }
KeyValueServiceCoordinationStore implements CoordinationStore<T> { @SuppressWarnings("OptionalUsedAsFieldOrParameterType") @VisibleForTesting CheckAndSetResult<SequenceAndBound> checkAndSetCoordinationValue( Optional<SequenceAndBound> oldValue, SequenceAndBound newValue) { if (oldValue.map(presentOldValue -> Objects.equals(presentOldValue, newValue)).orElse(false)) { return ImmutableCheckAndSetResult.of(true, ImmutableList.of(newValue)); } CheckAndSetRequest request = new CheckAndSetRequest.Builder() .table(AtlasDbConstants.COORDINATION_TABLE) .cell(getCoordinationValueCell()) .oldValue(oldValue.map(this::serializeSequenceAndBound)) .newValue(serializeSequenceAndBound(newValue)) .build(); try { kvs.checkAndSet(request); return ImmutableCheckAndSetResult.of(true, ImmutableList.of(newValue)); } catch (CheckAndSetException e) { if (e.getActualValues() != null) { return ImmutableCheckAndSetResult.of(false, e.getActualValues() .stream() .map(this::deserializeSequenceAndBound) .collect(Collectors.toList())); } return getFailedResultWithMostRecentValueAndBound(oldValue, newValue); } } @VisibleForTesting  KeyValueServiceCoordinationStore(
            ObjectMapper objectMapper,
            KeyValueService kvs,
            byte[] coordinationRow,
            LongSupplier sequenceNumberSupplier,
            BiPredicate<T, T> shouldReuseExtantValue,
            Class<T> clazz); static CoordinationStore<T> create(
            ObjectMapper objectMapper,
            KeyValueService kvs,
            byte[] coordinationRow,
            LongSupplier sequenceNumberSupplier,
            BiPredicate<T, T> shouldReuseExistingValue,
            Class<T> clazz,
            boolean initializeAsync); @Override boolean isInitialized(); @Override Optional<ValueAndBound<T>> getAgreedValue(); @Override CheckAndSetResult<ValueAndBound<T>> transformAgreedValue(Function<ValueAndBound<T>, T> transform); }
KeyValueServiceCoordinationStore implements CoordinationStore<T> { @SuppressWarnings("OptionalUsedAsFieldOrParameterType") @VisibleForTesting CheckAndSetResult<SequenceAndBound> checkAndSetCoordinationValue( Optional<SequenceAndBound> oldValue, SequenceAndBound newValue) { if (oldValue.map(presentOldValue -> Objects.equals(presentOldValue, newValue)).orElse(false)) { return ImmutableCheckAndSetResult.of(true, ImmutableList.of(newValue)); } CheckAndSetRequest request = new CheckAndSetRequest.Builder() .table(AtlasDbConstants.COORDINATION_TABLE) .cell(getCoordinationValueCell()) .oldValue(oldValue.map(this::serializeSequenceAndBound)) .newValue(serializeSequenceAndBound(newValue)) .build(); try { kvs.checkAndSet(request); return ImmutableCheckAndSetResult.of(true, ImmutableList.of(newValue)); } catch (CheckAndSetException e) { if (e.getActualValues() != null) { return ImmutableCheckAndSetResult.of(false, e.getActualValues() .stream() .map(this::deserializeSequenceAndBound) .collect(Collectors.toList())); } return getFailedResultWithMostRecentValueAndBound(oldValue, newValue); } } @VisibleForTesting  KeyValueServiceCoordinationStore(
            ObjectMapper objectMapper,
            KeyValueService kvs,
            byte[] coordinationRow,
            LongSupplier sequenceNumberSupplier,
            BiPredicate<T, T> shouldReuseExtantValue,
            Class<T> clazz); static CoordinationStore<T> create(
            ObjectMapper objectMapper,
            KeyValueService kvs,
            byte[] coordinationRow,
            LongSupplier sequenceNumberSupplier,
            BiPredicate<T, T> shouldReuseExistingValue,
            Class<T> clazz,
            boolean initializeAsync); @Override boolean isInitialized(); @Override Optional<ValueAndBound<T>> getAgreedValue(); @Override CheckAndSetResult<ValueAndBound<T>> transformAgreedValue(Function<ValueAndBound<T>, T> transform);  }