docs: private void createAckMailboxesIfNeeded(final Set<Integer> localPartitions) {
          m_mailboxesZKPath = VoltZK.exportGenerations + "/" + "mailboxes";
          if (m_mbox == null) {
              m_mbox = new LocalMailbox(m_messenger) {
                  @Override
                  public void deliver(VoltMessage message) {
                      if (message instanceof BinaryPayloadMessage) {
                          BinaryPayloadMessage bpm = (BinaryPayloadMessage)message;
                          ByteBuffer buf = ByteBuffer.wrap(bpm.m_payload);
                          final byte msgType = buf.get();
                          final int partition = buf.getInt();
                          final Map<String, ExportDataSource> partitionSources = m_dataSourcesByPartition.get(partition);
  
                          final int length = buf.getInt();
                          byte stringBytes[] = new byte[length];
                          buf.get(stringBytes);
                          String tableName = new String(stringBytes, Constants.UTF8ENCODING);
                          if (partitionSources == null) {
                              if (!m_removingPartitions.contains(partition)) {
                                  exportLog.error("Received an export message " + msgType + " for partition " + partition +
                                          " which does not exist on this node");
                              }
                              return;
                          }
                          final ExportDataSource eds = partitionSources.get(tableName);
                          if (eds == null) {
                              // For dangling buffers
                              if (msgType == ExportManager.TAKE_MASTERSHIP) {
                                  final long requestId = buf.getLong();
                                  if (exportLog.isDebugEnabled()) {
                                      exportLog.debug("Received TAKE_MASTERSHIP message(" + requestId +
                                              ") for a stream that no longer exists from " +
                                              CoreUtils.hsIdToString(message.m_sourceHSId) +
                                              " to " + CoreUtils.hsIdToString(m_mbox.getHSId()));
                                  }
                                  sendDummyTakeMastershipResponse(message.m_sourceHSId, requestId, partition, stringBytes);
                              } else {
                                  exportLog.warn("Received export message " + msgType + " for partition " +
                                          partition + " source " + tableName +
                                          " which does not exist on this node, sources = " + partitionSources);
                              }
                              return;
                          }
  
                          if (msgType == ExportManager.RELEASE_BUFFER) {
                              final long seqNo = buf.getLong();
                              final long catalogVersion = buf.getInt();
                              try {
                                  if (exportLog.isDebugEnabled()) {
                                      exportLog.debug("Received RELEASE_BUFFER message for " + eds.toString() +
                                              " , sequence number: " + seqNo + ", catalogVersion: " + catalogVersion +
                                              " from " + CoreUtils.hsIdToString(message.m_sourceHSId) +
                                              " to " + CoreUtils.hsIdToString(m_mbox.getHSId()));
                                  }
                                  eds.remoteAck(seqNo);
                              } catch (RejectedExecutionException ignoreIt) {
                                  // ignore it: as it is already shutdown
                              }
                          } else if (msgType == ExportManager.GIVE_MASTERSHIP) {
                              final long ackSeqNo = buf.getLong();
                              try {
                                  if (exportLog.isDebugEnabled()) {
                                      exportLog.debug("Received GIVE_MASTERSHIP message for " + eds.toString() +
                                              " with sequence number:" + ackSeqNo +
                                              " from " + CoreUtils.hsIdToString(message.m_sourceHSId) +
                                              " to " + CoreUtils.hsIdToString(m_mbox.getHSId()));
                                  }
                                  eds.remoteAck(ackSeqNo);
                              } catch (RejectedExecutionException ignoreIt) {
                                  // ignore it: as it is already shutdown
                              }
                              eds.acceptMastership();
                          } else if (msgType == ExportManager.GAP_QUERY) {
                              final long requestId = buf.getLong();
                              long gapStart = buf.getLong();
                              if (exportLog.isDebugEnabled()) {
                                  exportLog.debug("Received GAP_QUERY message(" + requestId +
                                          ") for " + eds.toString() +
                                          " from " + CoreUtils.hsIdToString(message.m_sourceHSId) +
                                          " to " + CoreUtils.hsIdToString(m_mbox.getHSId()));
                              }
                              eds.handleQueryMessage(message.m_sourceHSId, requestId, gapStart);
                          } else if (msgType == ExportManager.QUERY_RESPONSE) {
                              final long requestId = buf.getLong();
                              final long lastSeq = buf.getLong();
                              if (exportLog.isDebugEnabled()) {
                                  exportLog.debug("Received QUERY_RESPONSE message(" + requestId +
                                          "," + lastSeq + ") for " + eds.toString() +
                                          " from " + CoreUtils.hsIdToString(message.m_sourceHSId) +
                                          " to " + CoreUtils.hsIdToString(m_mbox.getHSId()));
                              }
                              eds.handleQueryResponse(message.m_sourceHSId, requestId, lastSeq);
                          } else if (msgType == ExportManager.TAKE_MASTERSHIP) {
                              final long requestId = buf.getLong();
                              if (exportLog.isDebugEnabled()) {
                                  exportLog.debug("Received TAKE_MASTERSHIP message(" + requestId +
                                          ") for " + eds.toString() +
                                          " from " + CoreUtils.hsIdToString(message.m_sourceHSId) +
                                          " to " + CoreUtils.hsIdToString(m_mbox.getHSId()));
                              }
                              eds.handleTakeMastershipMessage(message.m_sourceHSId, requestId);
                          } else if (msgType == ExportManager.TAKE_MASTERSHIP_RESPONSE) {
                              final long requestId = buf.getLong();
                              if (exportLog.isDebugEnabled()) {
                                  exportLog.debug("Received TAKE_MASTERSHIP_RESPONSE message(" + requestId +
                                          ") for " + eds.toString() +
                                          " from " + CoreUtils.hsIdToString(message.m_sourceHSId) +
                                          " to " + CoreUtils.hsIdToString(m_mbox.getHSId()));
                              }
                              eds.handleTakeMastershipResponse(message.m_sourceHSId, requestId);
                          } else {
                              exportLog.error("Receive unsupported message type " + message + " in export subsystem");
                          }
                      } else {
                          exportLog.error("Receive unexpected message " + message + " in export subsystem");
                      }
                  }
              };
              m_messenger.createMailbox(null, m_mbox);
          }
  
          // Rejoining node may receives gap query message before childUpdating thread gets back result,
          // in case it couldn't find local mailbox to send back response, add local mailbox to the list first.
          for (Integer partition : localPartitions) {
              updateAckMailboxes(partition, null);
          }
          // Update latest replica list to each data source.
          updateReplicaList(localPartitions);
      }