DistributedEventTracker implements EventTracker { @Override public void syncBulkOp(Runnable r, EventID eventID, boolean partOfTransaction) { if (partOfTransaction) { r.run(); return; } Assert.assertTrue(eventID != null); ThreadIdentifier membershipID = createThreadIDFromEvent(eventID); Object opSyncObj = null; do { opSyncObj = recordedBulkOps.putIfAbsent(membershipID, new Object()); if (opSyncObj == null) { opSyncObj = recordedBulkOps.get(membershipID); } } while (opSyncObj == null); synchronized (opSyncObj) { try { recordBulkOpStart(eventID, membershipID); r.run(); } finally { recordedBulkOps.remove(membershipID); } } } DistributedEventTracker(InternalCache cache, CancelCriterion stopper, String regionName); @Override void start(); @Override void stop(); @Override void clear(); @Override Map<ThreadIdentifier, EventSequenceNumberHolder> getState(); @Override void recordState(InternalDistributedMember provider,
      Map<ThreadIdentifier, EventSequenceNumberHolder> state); @Override void setInitialized(); @Override void waitOnInitialization(); @Override void recordEvent(InternalCacheEvent event); @Override boolean hasSeenEvent(InternalCacheEvent event); @Override boolean hasSeenEvent(EventID eventID); @Override boolean hasSeenEvent(EventID eventID, InternalCacheEvent tagHolder); @Override VersionTag findVersionTagForSequence(EventID eventID); @Override VersionTag findVersionTagForBulkOp(EventID eventID); @Override String getName(); @Override void syncBulkOp(Runnable r, EventID eventID, boolean partOfTransaction); @Override void recordBulkOpStart(EventID eventID, ThreadIdentifier tid); @Override boolean isInitialized(); @Override boolean isInitialImageProvider(DistributedMember mbr); @Override ConcurrentMap<ThreadIdentifier, BulkOperationHolder> getRecordedBulkOpVersionTags(); @Override ConcurrentMap<ThreadIdentifier, EventSequenceNumberHolder> getRecordedEvents(); @Override String toString();  }
@Test public void executesABulkOperations() { EventID eventID = new EventID(new byte[0], 0L, 1L); Runnable bulkOperation = mock(Runnable.class); eventTracker.syncBulkOp(bulkOperation, eventID, false); verify(bulkOperation, times(1)).run(); }
@Test public void executesRunnableIfNotPartOfATransaction() { EventID eventID = new EventID(new byte[0], 0L, 1L); Runnable bulkOperation = mock(Runnable.class); eventTracker.syncBulkOp(bulkOperation, eventID, true); verify(bulkOperation, times(1)).run(); }