private boolean putInBucket(final InternalDistributedMember targetNode, final Integer bucketId,
      final EntryEventImpl event, final boolean ifNew, boolean ifOld, Object expectedOldValue,
      boolean requireOldValue, final long lastModified) {
    if (logger.isDebugEnabled()) {
      logger.debug("putInBucket: {} ({}) to {} to bucketId={} retry={} ms", event.getKey(),
          event.getKey().hashCode(), targetNode, bucketStringForLogs(bucketId), retryTimeout);
    }
    // retry the put remotely until it finds the right node managing the bucket

    RetryTimeKeeper retryTime = null;
    boolean result = false;
    InternalDistributedMember currentTarget = targetNode;
    long timeOut = 0;
    int count = 0;
    for (;;) {
      switch (count) {
        case 0:
          // Note we don't check for DM cancellation in common case.
          // First time. Assume success, keep going.
          break;
        case 1:
          cache.getCancelCriterion().checkCancelInProgress(null);
          // Second time (first failure). Calculate timeout and keep going.
          timeOut = System.currentTimeMillis() + retryTimeout;
          break;
        default:
          cache.getCancelCriterion().checkCancelInProgress(null);
          // test for timeout
          long timeLeft = timeOut - System.currentTimeMillis();
          if (timeLeft < 0) {
            PRHARedundancyProvider.timedOut(this, null, null, "update an entry", retryTimeout);
            // NOTREACHED
          }

          // Didn't time out. Sleep a bit and then continue
          boolean interrupted = Thread.interrupted();
          try {
            Thread.sleep(PartitionedRegionHelper.DEFAULT_WAIT_PER_RETRY_ITERATION);
          } catch (InterruptedException ignore) {
            interrupted = true;
          } finally {
            if (interrupted) {
              Thread.currentThread().interrupt();
            }
          }
          break;
      } // switch
      count++;

      if (currentTarget == null) { // pick target
        checkReadiness();
        if (retryTime == null) {
          retryTime = new RetryTimeKeeper(retryTimeout);
        }
        currentTarget = waitForNodeOrCreateBucket(retryTime, event, bucketId);

        // It's possible this is a GemFire thread e.g. ServerConnection
        // which got to this point because of a distributed system shutdown or
        // region closure which uses interrupt to break any sleep() or wait() calls
        // e.g. waitForPrimary or waitForBucketRecovery in which case throw exception
        checkShutdown();
        continue;
      } // pick target

      try {
        final boolean isLocal = (localMaxMemory > 0) && currentTarget.equals(getMyId());
        if (logger.isDebugEnabled()) {
          logger.debug("putInBucket: currentTarget = {}; ifNew = {}; ifOld = {}; isLocal = {}",
              currentTarget, ifNew, ifOld, isLocal);
        }
        checkIfAboveThreshold(event);
        if (isLocal) {
          event.setInvokePRCallbacks(true);
          long start = prStats.startPutLocal();
          try {
            final BucketRegion br =
                dataStore.getInitializedBucketForId(event.getKey(), bucketId);
            // Local updates should insert a serialized (aka CacheDeserializable) object
            // given that most manipulation of values is remote (requiring serialization to send).
            // But... function execution always implies local manipulation of
            // values so keeping locally updated values in Object form should be more efficient.
            if (!FunctionExecutionPooledExecutor.isFunctionExecutionThread()) {
              // TODO: this condition may not help since BucketRegion.virtualPut calls
              // forceSerialized
              br.forceSerialized(event);
            }
            if (ifNew) {
              result = dataStore.createLocally(br, event, ifNew, ifOld, requireOldValue,
                  lastModified);
            } else {
              result = dataStore.putLocally(br, event, ifNew, ifOld, expectedOldValue,
                  requireOldValue, lastModified);
            }
          } finally {
            prStats.endPutLocal(start);
          }
        } // local
        else { // remote
          // no need to perform early serialization (and create an un-necessary byte array)
          // sending the message performs that work.
          long start = prStats.startPutRemote();
          try {
            if (ifNew) {
              result = createRemotely(currentTarget, bucketId, event, requireOldValue);
            } else {
              result = putRemotely(currentTarget, event, ifNew, ifOld, expectedOldValue,
                  requireOldValue);
              if (!requireOldValue) {
                // make sure old value is set to NOT_AVAILABLE token
                event.oldValueNotAvailable();
              }
            }
          } finally {
            prStats.endPutRemote(start);
          }
        } // remote

        if (!result && !ifOld && !ifNew) {
          Assert.assertTrue(!isLocal);
          ForceReattemptException fre = new ForceReattemptException(
              "false result when !ifNew and !ifOld is unacceptable - retrying");
          fre.setHash(event.getKey().hashCode());
          throw fre;
        }

        return result;
      } catch (ConcurrentCacheModificationException e) {
        if (logger.isDebugEnabled()) {
          logger.debug("putInBucket: caught concurrent cache modification exception", e);
        }
        event.isConcurrencyConflict(true);

        if (logger.isTraceEnabled()) {
          logger.trace(
              "ConcurrentCacheModificationException received for putInBucket for bucketId: {}{}{} for event: {}  No reattampt is done, returning from here",
              getPRId(), BUCKET_ID_SEPARATOR, bucketId, event);
        }
        return result;
      } catch (ForceReattemptException prce) {
        prce.checkKey(event.getKey());
        if (logger.isDebugEnabled()) {
          logger.debug(
              "putInBucket: Got ForceReattemptException for {} on VM {} for node {}{}{} for bucket = {}",
              this, getMyId(), currentTarget, getPRId(), BUCKET_ID_SEPARATOR, bucketId, prce);
          logger.debug("putInBucket: count={}", count);
        }
        checkReadiness();
        InternalDistributedMember lastTarget = currentTarget;
        if (retryTime == null) {
          retryTime = new RetryTimeKeeper(retryTimeout);
        }
        currentTarget = getNodeForBucketWrite(bucketId, retryTime);
        if (lastTarget.equals(currentTarget)) {
          if (retryTime.overMaximum()) {
            PRHARedundancyProvider.timedOut(this, null, null, "update an entry", retryTimeout);
            // NOTREACHED
          }
          retryTime.waitToRetryNode();
        }
        event.setPossibleDuplicate(true);
      } catch (PrimaryBucketException notPrimary) {
        if (logger.isDebugEnabled()) {
          logger.debug("Bucket {} on Node {} not primary", notPrimary.getLocalizedMessage(),
              currentTarget);
        }
        getRegionAdvisor().notPrimary(bucketId, currentTarget);
        if (retryTime == null) {
          retryTime = new RetryTimeKeeper(retryTimeout);
        }
        currentTarget = getNodeForBucketWrite(bucketId, retryTime);
      }

      // It's possible this is a GemFire thread e.g. ServerConnection
      // which got to this point because of a distributed system shutdown or
      // region closure which uses interrupt to break any sleep() or wait()
      // calls
      // e.g. waitForPrimary or waitForBucketRecovery in which case throw
      // exception
      checkShutdown();

      // If we get here, the attempt failed...
      if (count == 1) {
        if (ifNew) {
          prStats.incCreateOpsRetried();
        } else {
          prStats.incPutOpsRetried();
        }
      }
      if (event.getOperation().isCreate()) {
        prStats.incCreateRetries();
      } else {
        prStats.incPutRetries();
      }

      if (logger.isDebugEnabled()) {
        logger.debug(
            "putInBucket for bucketId = {} failed (attempt # {} ({} ms left), retrying with node {}",
            bucketStringForLogs(bucketId), count, (timeOut - System.currentTimeMillis()),
            currentTarget);
      }
    } // for

    // NOTREACHED
  }
Performs the {@link Operation#UPDATE put} operation in the bucket on the remote VM and invokes
"put" callbacks

@param targetNode the VM in whose {@link PartitionedRegionDataStore} contains the bucket
@param bucketId the identity of the bucket
@param event the event that contains the key and value
@param lastModified the timestamp that the entry was modified
@param ifNew true=can create a new key false=can't create a new key
@param ifOld true=can update existing entry false=can't update existing entry
@param expectedOldValue only succeed if old value is equal to this value. If null, then doesn't
       matter what old value is. If INVALID token, must be INVALID.
@see LocalRegion#virtualPut(EntryEventImpl, boolean, boolean, Object, boolean, long, boolean)
@return false if ifNew is true and there is an existing key, or ifOld is true and there is no
        existing entry; otherwise return true.
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java