public void process(LocalRegion lr, DistributedMember src, long lastMod) {
    if (op.isRegion()) {
      // it is a region operation
      RegionEventImpl re = new RegionEventImpl(lr, op, cbArg, true, src);
      // re.setQueued(true);
      if (op.isRegionInvalidate()) {
        lr.basicInvalidateRegion(re);
      } else if (op == Operation.REGION_CLEAR) {
        lr.cmnClearRegion(re, false/* cacheWrite */, false/* useRVV */);
      } else {
        throw new IllegalStateException(
            String.format("The %s should not have been queued.",
                op));
      }
    } else {
      // it is an entry operation
      // TODO :EventID should be passed from the sender & should be reused here
      @Released
      EntryEventImpl ee = EntryEventImpl.create(lr, op, key, null, cbArg, true, src);
      try {
        // ee.setQueued(true);
        if (op.isCreate() || op.isUpdate()) {
          UpdateOperation.UpdateMessage.setNewValueInEvent(value, valueObj, ee,
              deserializationPolicy);
          try {
            long time = lastMod;
            if (ee.getVersionTag() != null) {
              time = ee.getVersionTag().getVersionTimeStamp();
            }
            if (AbstractUpdateOperation.doPutOrCreate(lr, ee, time)) {
              // am I done?
            }
          } catch (ConcurrentCacheModificationException e) {
            // operation was rejected by the cache's concurrency control mechanism as being old
          }
        } else if (op.isDestroy()) {
          ee.setOldValueFromRegion();
          try {
            lr.basicDestroy(ee, false, null); // expectedOldValue
                                              // ???:ezoerner:20080815
                                              // can a remove(key, value) operation be
                                              // queued?
          } catch (ConcurrentCacheModificationException e) {
            // operation was rejected by the cache's concurrency control mechanism as being old
          } catch (EntryNotFoundException ignore) {
          } catch (CacheWriterException e) {
            throw new Error("CacheWriter should not be called", e);
          } catch (TimeoutException e) {
            throw new Error("DistributedLock should not be acquired", e);
          }
        } else if (op.isInvalidate()) {
          ee.setOldValueFromRegion();
          boolean forceNewEntry = lr.getDataPolicy().withReplication() && !lr.isInitialized();
          boolean invokeCallbacks = lr.isInitialized();
          try {
            lr.basicInvalidate(ee, invokeCallbacks, forceNewEntry);
          } catch (ConcurrentCacheModificationException e) {
            // operation was rejected by the cache's concurrency control mechanism as being old
          } catch (EntryNotFoundException ignore) {
          }
        } else {
          throw new IllegalStateException(
              String.format("The %s should not have been queued.",
                  op));
        }
      } finally {
        ee.release();
      }
    }
  }
Creates an instance of a queued operation with the given data.
geode-core/src/main/java/org/apache/geode/internal/cache/QueuedOperation.java