private void send() {
    // We should check if there are normal replicas first
    int toSendReplicaCount = regionReplication - 1 - failedReplicas.size();
    if (toSendReplicaCount <= 0) {
      LOG.warn("All replicas {} are failed, exit send....", failedReplicas);
      return;
    }

    List<SinkEntry> toSend = new ArrayList<>();
    long totalSize = 0L;
    boolean hasMetaEdit = false;
    for (SinkEntry entry;;) {
      entry = entries.poll();
      if (entry == null) {
        break;
      }
      toSend.add(entry);
      totalSize += entry.size;
      hasMetaEdit |= entry.edit.isMetaEdit();
      if (toSend.size() >= batchCountCapacity || totalSize >= batchSizeCapacity) {
        break;
      }
    }

    long rpcTimeoutNsToUse;
    long operationTimeoutNsToUse;
    if (!hasMetaEdit) {
      rpcTimeoutNsToUse = rpcTimeoutNs;
      operationTimeoutNsToUse = operationTimeoutNs;
    } else {
      rpcTimeoutNsToUse = metaEditRpcTimeoutNs;
      operationTimeoutNsToUse = metaEditOperationTimeoutNs;
    }
    sending = true;
    List<WAL.Entry> walEntries =
      toSend.stream().map(e -> new WAL.Entry(e.key, e.edit)).collect(Collectors.toList());
    AtomicInteger remaining = new AtomicInteger(toSendReplicaCount);
    Map<Integer, MutableObject<Throwable>> replica2Error = new HashMap<>();
    for (int replicaId = 1; replicaId < regionReplication; replicaId++) {
      if (failedReplicas.contains(replicaId)) {
        continue;
      }
      MutableObject<Throwable> error = new MutableObject<>();
      replica2Error.put(replicaId, error);
      RegionInfo replica = RegionReplicaUtil.getRegionInfoForReplica(primary, replicaId);
      FutureUtils.addListener(
        conn.replicate(replica, walEntries, retries, rpcTimeoutNsToUse, operationTimeoutNsToUse),
        (r, e) -> {
          error.setValue(e);
          if (remaining.decrementAndGet() == 0) {
            onComplete(toSend, replica2Error);
          }
        });
    }
  }
Should be called regardless of the result of the replicating operation. Unless you still want
to reuse this entry, otherwise you must call this method to release the possible off heap
memories.
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/regionreplication/RegionReplicationSink.java