@Override
    public void process(final DistributionMessage msg) {

      if (msg instanceof ReplyMessage) {
        ReplyException ex = ((ReplyMessage) msg).getException();
        if (putMessage.bridgeContext == null
            // Why is this code not happening for bug 41916?
            && (ex != null && ex.getCause() instanceof InvalidDeltaException)) {
          final PutMessage putMsg = new PutMessage(putMessage);
          final DistributionManager dm = getDistributionManager();
          Runnable sendFullObject = new Runnable() {
            @Override
            public void run() {
              putMsg.resetRecipients();
              putMsg.setRecipient(msg.getSender());
              putMsg.setSendDelta(false);
              if (logger.isDebugEnabled()) {
                logger.debug("Sending full object({}) to {}", putMsg,
                    putMsg.getRecipientsDescription());
              }
              dm.putOutgoing(putMsg);

              // Update stats
              try {
                PartitionedRegion.getPRFromId(putMsg.regionId).getCachePerfStats()
                    .incDeltaFullValuesSent();
              } catch (Exception ignored) {
              }
            }

            @Override
            public String toString() {
              return "Sending full object {" + putMsg + "}";
            }
          };
          if (isExpectingDirectReply()) {
            sendFullObject.run();
          } else {
            getDistributionManager().getExecutors().getWaitingThreadPool().execute(sendFullObject);
          }
          return;
        }
      }
      super.process(msg);
    }
@return the result of the remote put operation
@throws ForceReattemptException if the peer is no longer available
@throws CacheException if the peer generates an error
geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/PutMessage.java