RepairEntryProcessor implements EntryProcessor { @SuppressWarnings("unchecked") @Override public Object process(MutableEntry entry, Object... arguments) throws EntryProcessorException { GridCacheContext cctx = cacheContext(entry); GridCacheVersion currKeyGridCacheVer = keyVersion(entry); if (topologyChanged(cctx, startTopVer)) throw new EntryProcessorException("Topology version was changed"); UUID locNodeId = cctx.localNodeId(); VersionedValue versionedVal = data.get(locNodeId); if (versionedVal != null) { if (currKeyGridCacheVer.compareTo(versionedVal.version()) == 0) { if (val == null) entry.remove(); else entry.setValue(val); return RepairStatus.SUCCESS; } else return RepairStatus.CONCURRENT_MODIFICATION; } else { if (currKeyGridCacheVer.compareTo(new GridCacheVersion(0, 0, 0)) == 0) { long recheckStartTime = minValue(VersionedValue::recheckStartTime); boolean inEntryTTLBounds = (System.currentTimeMillis() - recheckStartTime) < Long.getLong(IGNITE_CACHE_REMOVED_ENTRIES_TTL, 10_000); long minUpdateCntr = minValue(VersionedValue::updateCounter); long currUpdateCntr = updateCounter(cctx, entry.getKey()); boolean inDeferredDelQueueBounds = ((currUpdateCntr - minUpdateCntr) < rmvQueueMaxSize); if (cctx.config().getAtomicityMode() != CacheAtomicityMode.ATOMIC || inEntryTTLBounds && inDeferredDelQueueBounds) { if (val == null) entry.remove(); else entry.setValue(val); return RepairStatus.SUCCESS; } } else return RepairStatus.CONCURRENT_MODIFICATION; if (forceRepair) { if (val == null) entry.remove(); else entry.setValue(val); return RepairStatus.SUCCESS; } return RepairStatus.FAIL; } } @SuppressWarnings("AssignmentOrReturnOfFieldWithMutableType")  RepairEntryProcessor(
        Object val,
        Map<UUID, VersionedValue> data,
        long rmvQueueMaxSize,
        boolean forceRepair,
        AffinityTopologyVersion startTopVer); @SuppressWarnings("unchecked") @Override Object process(MutableEntry entry, Object... arguments);  }
@Test public void testForceRepairApplyRemove() { final boolean forceRepair = true; Map<UUID, VersionedValue> data = new HashMap<>(); data.put(OTHRER_NODE_ID, new VersionedValue( OLD_CACHE_VALUE, new GridCacheVersion(1, 1, 1), 1, 1 )); RepairEntryProcessor repairProcessor = new RepairEntryProcessorStub( null, data, RMV_QUEUE_MAX_SIZE, forceRepair, new AffinityTopologyVersion(1) ).setKeyVersion(new GridCacheVersion(0, 0, 0)); MutableEntry entry = mock(MutableEntry.class); assertEquals(repairProcessor.process(entry), RepairEntryProcessor.RepairStatus.SUCCESS); verify(entry, times(1)).remove(); }
@Test public void testForceRepairApplyValue() { final boolean forceRepair = true; Map<UUID, VersionedValue> data = new HashMap<>(); data.put(OTHRER_NODE_ID, new VersionedValue( OLD_CACHE_VALUE, new GridCacheVersion(1, 1, 1), 1, 1 )); RepairEntryProcessor repairProcessor = new RepairEntryProcessorStub( NEW_VALUE, data, RMV_QUEUE_MAX_SIZE, forceRepair, new AffinityTopologyVersion(1) ).setKeyVersion(new GridCacheVersion(0, 0, 0)); MutableEntry entry = mock(MutableEntry.class); assertEquals(repairProcessor.process(entry), RepairEntryProcessor.RepairStatus.SUCCESS); verify(entry, times(1)).setValue(NEW_VALUE); }
@Test public void testSetValueWithoutParallelUpdateWhereCurrentRecheckNotNull() { Map<UUID, VersionedValue> data = new HashMap<>(); data.put(LOCAL_NODE_ID, new VersionedValue( OLD_CACHE_VALUE, new GridCacheVersion(1, 1, 1), 1, 1 )); RepairEntryProcessor repairProcessor = new RepairEntryProcessorStub( NEW_VALUE, data, RMV_QUEUE_MAX_SIZE, false, new AffinityTopologyVersion(1) ).setKeyVersion(new GridCacheVersion(1, 1, 1)); MutableEntry entry = mock(MutableEntry.class); when(entry.getValue()).thenReturn(OLD_CACHE_VALUE); assertEquals(repairProcessor.process(entry), RepairEntryProcessor.RepairStatus.SUCCESS); verify(entry, times(1)).setValue(NEW_VALUE); }
@Test public void testRemoveValueWithoutParallelUpdateWhereCurrentRecheckNotNull() { Map<UUID, VersionedValue> data = new HashMap<>(); data.put(LOCAL_NODE_ID, new VersionedValue( OLD_CACHE_VALUE, new GridCacheVersion(1, 1, 1), 1, 1 )); RepairEntryProcessor repairProcessor = new RepairEntryProcessorStub( null, data, RMV_QUEUE_MAX_SIZE, false, new AffinityTopologyVersion(1) ).setKeyVersion(new GridCacheVersion(1, 1, 1)); MutableEntry entry = mock(MutableEntry.class); when(entry.getValue()).thenReturn(OLD_CACHE_VALUE); assertEquals(repairProcessor.process(entry), RepairEntryProcessor.RepairStatus.SUCCESS); verify(entry, times(1)).remove(); }
@Test public void testWriteValueParallelUpdateWhereCurrentRecheckNotNull() { Map<UUID, VersionedValue> data = new HashMap<>(); data.put(LOCAL_NODE_ID, new VersionedValue( OLD_CACHE_VALUE, new GridCacheVersion(1, 1, 1), 1, 1 )); RepairEntryProcessor repairProcessor = new RepairEntryProcessorStub( NEW_VALUE, data, RMV_QUEUE_MAX_SIZE, false, new AffinityTopologyVersion(1) ).setKeyVersion(new GridCacheVersion(2, 2, 2)); MutableEntry entry = mock(MutableEntry.class); when(entry.getValue()).thenReturn(OLD_CACHE_VALUE); assertEquals(repairProcessor.process(entry), RepairEntryProcessor.RepairStatus.CONCURRENT_MODIFICATION); }
@Test public void testRecheckVersionNullCurrentValueExist() { RepairEntryProcessor repairProcessor = new RepairEntryProcessorStub( NEW_VALUE, new HashMap<>(), RMV_QUEUE_MAX_SIZE, false, new AffinityTopologyVersion(1) ).setKeyVersion(new GridCacheVersion(1, 1, 1)); MutableEntry entry = mock(MutableEntry.class); assertEquals(repairProcessor.process(entry), RepairEntryProcessor.RepairStatus.CONCURRENT_MODIFICATION); }
@Test public void testRecheckVersionNullAndTtlEntryShouldNotAlreadyRemovedAndNewUpdateCounterLessDelQueueSizeOpRemove() { Map<UUID, VersionedValue> data = new HashMap<>(); data.put(OTHRER_NODE_ID, new VersionedValue( OLD_CACHE_VALUE, new GridCacheVersion(1, 1, 1), 1, System.currentTimeMillis() )); RepairEntryProcessor repairProcessor = new RepairEntryProcessorStub( null, data, RMV_QUEUE_MAX_SIZE, false, new AffinityTopologyVersion(1) ).setKeyVersion(new GridCacheVersion(0, 0, 0)); MutableEntry entry = mock(MutableEntry.class); assertEquals(repairProcessor.process(entry), RepairEntryProcessor.RepairStatus.SUCCESS); verify(entry, times(1)).remove(); }
@Test public void testRecheckVersionNullAndTtlEntryShouldNotAlreadyRemovedAndNewUpdateCounterLessDelQueueSizeOpSet() { Map<UUID, VersionedValue> data = new HashMap<>(); data.put(OTHRER_NODE_ID, new VersionedValue( OLD_CACHE_VALUE, new GridCacheVersion(1, 1, 1), 1, System.currentTimeMillis() )); RepairEntryProcessor repairProcessor = new RepairEntryProcessorStub( RECHECK_VALUE, data, RMV_QUEUE_MAX_SIZE, false, new AffinityTopologyVersion(1) ).setKeyVersion(new GridCacheVersion(0, 0, 0)); MutableEntry entry = mock(MutableEntry.class); assertEquals(repairProcessor.process(entry), RepairEntryProcessor.RepairStatus.SUCCESS); verify(entry, times(1)).setValue(RECHECK_VALUE); }
@Test public void testEntryWasChangedDuringRepairAtOtherValue() { Map<UUID, VersionedValue> data = new HashMap<>(); data.put(LOCAL_NODE_ID, new VersionedValue( new CacheObjectImpl(OLD_VALUE, OLD_VALUE.getBytes()), new GridCacheVersion(1, 1, 1), 1, 1 )); RepairEntryProcessor repairProcessor = new RepairEntryProcessorStub( null, data, RMV_QUEUE_MAX_SIZE, false, new AffinityTopologyVersion(1) ).setKeyVersion(new GridCacheVersion(0, 0, 0)); MutableEntry entry = mock(MutableEntry.class); when(entry.getValue()).thenReturn(RECHECK_CACHE_VALUE); assertEquals(repairProcessor.process(entry), RepairEntryProcessor.RepairStatus.CONCURRENT_MODIFICATION); }
@Test public void testEntryWasChangedDuringRepairAtNull() { Map<UUID, VersionedValue> data = new HashMap<>(); data.put(LOCAL_NODE_ID, new VersionedValue( new CacheObjectImpl(OLD_VALUE, OLD_VALUE.getBytes()), new GridCacheVersion(1, 1, 1), 1, 1 )); RepairEntryProcessor repairProcessor = new RepairEntryProcessorStub( null, data, RMV_QUEUE_MAX_SIZE, false, new AffinityTopologyVersion(1) ).setKeyVersion(new GridCacheVersion(0, 0, 0)); MutableEntry entry = mock(MutableEntry.class); when(entry.getValue()).thenReturn(null); assertEquals(repairProcessor.process(entry), RepairEntryProcessor.RepairStatus.CONCURRENT_MODIFICATION); }
@Test public void testEntryWasChangedDuringRepairFromNullToValue() { Map<UUID, VersionedValue> data = new HashMap<>(); data.put(OTHRER_NODE_ID, new VersionedValue( OLD_CACHE_VALUE, new GridCacheVersion(1, 1, 1), 1, 1 )); RepairEntryProcessor repairProcessor = new RepairEntryProcessorStub( null, data, RMV_QUEUE_MAX_SIZE, false, new AffinityTopologyVersion(1) ).setKeyVersion(new GridCacheVersion(1, 1, 1)); MutableEntry entry = mock(MutableEntry.class); when(entry.getValue()).thenReturn(new CacheObjectImpl(RECHECK_VALUE, RECHECK_VALUE.getBytes())); assertEquals(repairProcessor.process(entry), RepairEntryProcessor.RepairStatus.CONCURRENT_MODIFICATION); }
@Test public void testRecheckVersionNullAndTtlEntryExpired() { Map<UUID, VersionedValue> data = new HashMap<>(); data.put(OTHRER_NODE_ID, new VersionedValue( new CacheObjectImpl(OLD_VALUE, OLD_VALUE.getBytes()), new GridCacheVersion(1, 1, 1), 1, 1 )); RepairEntryProcessor repairProcessor = new RepairEntryProcessorStub( null, data, RMV_QUEUE_MAX_SIZE, false, new AffinityTopologyVersion(1) ).setKeyVersion(new GridCacheVersion(0, 0, 0)); MutableEntry entry = mock(MutableEntry.class); assertEquals(repairProcessor.process(entry), RepairEntryProcessor.RepairStatus.FAIL); }
@Test public void testRecheckVersionNullAndDefDelQueueExpired() { Map<UUID, VersionedValue> data = new HashMap<>(); data.put(OTHRER_NODE_ID, new VersionedValue( new CacheObjectImpl(OLD_VALUE, OLD_VALUE.getBytes()), new GridCacheVersion(1, 1, 1), 1, 1 )); RepairEntryProcessor repairProcessor = new RepairEntryProcessorStub( null, data, RMV_QUEUE_MAX_SIZE, false, new AffinityTopologyVersion(1) ) .setKeyVersion(new GridCacheVersion(0, 0, 0)) .setUpdateCounter(100); MutableEntry entry = mock(MutableEntry.class); assertEquals(repairProcessor.process(entry), RepairEntryProcessor.RepairStatus.FAIL); }