Committed by
Gerrit Code Review
Support null timestamps during EC Map remove
Change-Id: I250cc08d6b2570fd9febe5fc50ab0556bedfa410
Showing
2 changed files
with
23 additions
and
15 deletions
... | @@ -330,10 +330,13 @@ public class EventuallyConsistentMapImpl<K, V> | ... | @@ -330,10 +330,13 @@ public class EventuallyConsistentMapImpl<K, V> |
330 | } | 330 | } |
331 | 331 | ||
332 | private V removeAndNotify(K key, V value) { | 332 | private V removeAndNotify(K key, V value) { |
333 | - MapValue<V> tombstone = MapValue.tombstone(timestampProvider.apply(key, value)); | 333 | + Timestamp timestamp = timestampProvider.apply(key, value); |
334 | + Optional<MapValue<V>> tombstone = tombstonesDisabled || timestamp == null | ||
335 | + ? Optional.empty() : Optional.of(MapValue.tombstone(timestamp)); | ||
334 | MapValue<V> previousValue = removeInternal(key, Optional.ofNullable(value), tombstone); | 336 | MapValue<V> previousValue = removeInternal(key, Optional.ofNullable(value), tombstone); |
335 | if (previousValue != null) { | 337 | if (previousValue != null) { |
336 | - notifyPeers(new UpdateEntry<>(key, tombstone), peerUpdateFunction.apply(key, previousValue.get())); | 338 | + notifyPeers(new UpdateEntry<>(key, tombstone.orElse(null)), |
339 | + peerUpdateFunction.apply(key, previousValue.get())); | ||
337 | if (previousValue.isAlive()) { | 340 | if (previousValue.isAlive()) { |
338 | notifyListeners(new EventuallyConsistentMapEvent<>(mapName, REMOVE, key, previousValue.get())); | 341 | notifyListeners(new EventuallyConsistentMapEvent<>(mapName, REMOVE, key, previousValue.get())); |
339 | } | 342 | } |
... | @@ -341,11 +344,11 @@ public class EventuallyConsistentMapImpl<K, V> | ... | @@ -341,11 +344,11 @@ public class EventuallyConsistentMapImpl<K, V> |
341 | return previousValue != null ? previousValue.get() : null; | 344 | return previousValue != null ? previousValue.get() : null; |
342 | } | 345 | } |
343 | 346 | ||
344 | - private MapValue<V> removeInternal(K key, Optional<V> value, MapValue<V> tombstone) { | 347 | + private MapValue<V> removeInternal(K key, Optional<V> value, Optional<MapValue<V>> tombstone) { |
345 | checkState(!destroyed, destroyedMessage); | 348 | checkState(!destroyed, destroyedMessage); |
346 | checkNotNull(key, ERROR_NULL_KEY); | 349 | checkNotNull(key, ERROR_NULL_KEY); |
347 | checkNotNull(value, ERROR_NULL_VALUE); | 350 | checkNotNull(value, ERROR_NULL_VALUE); |
348 | - checkState(tombstone.isTombstone()); | 351 | + tombstone.ifPresent(v -> checkState(v.isTombstone())); |
349 | 352 | ||
350 | counter.incrementCount(); | 353 | counter.incrementCount(); |
351 | AtomicBoolean updated = new AtomicBoolean(false); | 354 | AtomicBoolean updated = new AtomicBoolean(false); |
... | @@ -358,22 +361,26 @@ public class EventuallyConsistentMapImpl<K, V> | ... | @@ -358,22 +361,26 @@ public class EventuallyConsistentMapImpl<K, V> |
358 | if (existing == null) { | 361 | if (existing == null) { |
359 | log.debug("ECMap Remove: Existing value for key {} is already null", k); | 362 | log.debug("ECMap Remove: Existing value for key {} is already null", k); |
360 | } | 363 | } |
361 | - updated.set(valueMatches && (existing == null || tombstone.isNewerThan(existing))); | 364 | + if (valueMatches) { |
362 | - if (updated.get()) { | 365 | + if (existing == null) { |
363 | - previousValue.set(existing); | 366 | + updated.set(tombstone.isPresent()); |
367 | + } else { | ||
368 | + updated.set(!tombstone.isPresent() || tombstone.get().isNewerThan(existing)); | ||
369 | + } | ||
364 | } | 370 | } |
365 | if (updated.get()) { | 371 | if (updated.get()) { |
366 | - return tombstonesDisabled ? null : tombstone; | 372 | + previousValue.set(existing); |
373 | + return tombstone.orElse(null); | ||
367 | } else { | 374 | } else { |
368 | return existing; | 375 | return existing; |
369 | } | 376 | } |
370 | }); | 377 | }); |
371 | if (updated.get()) { | 378 | if (updated.get()) { |
372 | if (persistent) { | 379 | if (persistent) { |
373 | - if (tombstonesDisabled) { | 380 | + if (tombstone.isPresent()) { |
374 | - persistentStore.remove(key); | 381 | + persistentStore.update(key, tombstone.get()); |
375 | } else { | 382 | } else { |
376 | - persistentStore.update(key, tombstone); | 383 | + persistentStore.remove(key); |
377 | } | 384 | } |
378 | } | 385 | } |
379 | } | 386 | } |
... | @@ -605,9 +612,10 @@ public class EventuallyConsistentMapImpl<K, V> | ... | @@ -605,9 +612,10 @@ public class EventuallyConsistentMapImpl<K, V> |
605 | if (remoteValueDigest != null | 612 | if (remoteValueDigest != null |
606 | && remoteValueDigest.isNewerThan(localValue.digest()) | 613 | && remoteValueDigest.isNewerThan(localValue.digest()) |
607 | && remoteValueDigest.isTombstone()) { | 614 | && remoteValueDigest.isTombstone()) { |
615 | + MapValue<V> tombstone = MapValue.tombstone(remoteValueDigest.timestamp()); | ||
608 | MapValue<V> previousValue = removeInternal(key, | 616 | MapValue<V> previousValue = removeInternal(key, |
609 | Optional.empty(), | 617 | Optional.empty(), |
610 | - MapValue.tombstone(remoteValueDigest.timestamp())); | 618 | + Optional.of(tombstone)); |
611 | if (previousValue != null && previousValue.isAlive()) { | 619 | if (previousValue != null && previousValue.isAlive()) { |
612 | externalEvents.add(new EventuallyConsistentMapEvent<>(mapName, REMOVE, key, previousValue.get())); | 620 | externalEvents.add(new EventuallyConsistentMapEvent<>(mapName, REMOVE, key, previousValue.get())); |
613 | } | 621 | } |
... | @@ -623,8 +631,8 @@ public class EventuallyConsistentMapImpl<K, V> | ... | @@ -623,8 +631,8 @@ public class EventuallyConsistentMapImpl<K, V> |
623 | updates.forEach(update -> { | 631 | updates.forEach(update -> { |
624 | final K key = update.key(); | 632 | final K key = update.key(); |
625 | final MapValue<V> value = update.value(); | 633 | final MapValue<V> value = update.value(); |
626 | - if (value.isTombstone()) { | 634 | + if (value == null || value.isTombstone()) { |
627 | - MapValue<V> previousValue = removeInternal(key, Optional.empty(), value); | 635 | + MapValue<V> previousValue = removeInternal(key, Optional.empty(), Optional.ofNullable(value)); |
628 | if (previousValue != null && previousValue.isAlive()) { | 636 | if (previousValue != null && previousValue.isAlive()) { |
629 | notifyListeners(new EventuallyConsistentMapEvent<>(mapName, REMOVE, key, previousValue.get())); | 637 | notifyListeners(new EventuallyConsistentMapEvent<>(mapName, REMOVE, key, previousValue.get())); |
630 | } | 638 | } | ... | ... |
... | @@ -34,7 +34,7 @@ final class UpdateEntry<K, V> { | ... | @@ -34,7 +34,7 @@ final class UpdateEntry<K, V> { |
34 | */ | 34 | */ |
35 | public UpdateEntry(K key, MapValue<V> value) { | 35 | public UpdateEntry(K key, MapValue<V> value) { |
36 | this.key = checkNotNull(key); | 36 | this.key = checkNotNull(key); |
37 | - this.value = checkNotNull(value); | 37 | + this.value = value; |
38 | } | 38 | } |
39 | 39 | ||
40 | /** | 40 | /** | ... | ... |
-
Please register or login to post a comment