docs: public void map(byte[] keyBytes, byte[] valBytes, AbstractCollectorWrapper collector) throws IOException {
          // Compress key and values if required
          if(keySerializerDefinition.hasCompression()) {
              keyBytes = keyCompressor.deflate(keyBytes);
          }
  
          if(valueSerializerDefinition.hasCompression()) {
              valBytes = valueCompressor.deflate(valBytes);
          }
  
          // Get the output byte arrays ready to populate
          byte[] outputValue;
          byte[] outputKey;
  
          // Leave initial offset for (a) node id (b) partition id
          // since they are written later
          int offsetTillNow = 2 * ByteUtils.SIZE_OF_INT;
  
          if(getSaveKeys()) {
  
              // In order - 4 ( for node id ) + 4 ( partition id ) + 1 ( replica
              // type - primary | secondary | tertiary... ] + 4 ( key size )
              // size ) + 4 ( value size ) + key + value
              outputValue = new byte[valBytes.length + keyBytes.length + ByteUtils.SIZE_OF_BYTE + 4
                      * ByteUtils.SIZE_OF_INT];
  
              // Write key length - leave byte for replica type
              offsetTillNow += ByteUtils.SIZE_OF_BYTE;
              ByteUtils.writeInt(outputValue, keyBytes.length, offsetTillNow);
  
              // Write value length
              offsetTillNow += ByteUtils.SIZE_OF_INT;
              ByteUtils.writeInt(outputValue, valBytes.length, offsetTillNow);
  
              // Write key
              offsetTillNow += ByteUtils.SIZE_OF_INT;
              System.arraycopy(keyBytes, 0, outputValue, offsetTillNow, keyBytes.length);
  
              // Write value
              offsetTillNow += keyBytes.length;
              System.arraycopy(valBytes, 0, outputValue, offsetTillNow, valBytes.length);
  
              // Generate MR key - upper 8 bytes of 16 byte md5
              outputKey = ByteUtils.copy(md5er.digest(keyBytes),
                      0,
                      2 * ByteUtils.SIZE_OF_INT);
  
          } else {
  
              // In order - 4 ( for node id ) + 4 ( partition id ) + value
              outputValue = new byte[valBytes.length + 2 * ByteUtils.SIZE_OF_INT];
  
              // Write value
              System.arraycopy(valBytes, 0, outputValue, offsetTillNow, valBytes.length);
  
              // Generate MR key - 16 byte md5
              outputKey = md5er.digest(keyBytes);
  
          }
  
          // Generate partition and node list this key is destined for
          List<Integer> partitionList = routingStrategy.getPartitionList(keyBytes);
          Node[] partitionToNode = routingStrategy.getPartitionToNode();
  
          // In buildPrimaryReplicasOnly mode, we want to push out no more than a single replica
          // for each key. Otherwise (in vintage mode), we push out one copy per replica.
          int numberOfReplicasToPushTo = getBuildPrimaryReplicasOnly() ? 1 : partitionList.size();
  
          for(int replicaType = 0; replicaType < numberOfReplicasToPushTo; replicaType++) {
  
              // Node id
              ByteUtils.writeInt(outputValue,
                                 partitionToNode[partitionList.get(replicaType)].getId(),
                                 0);
  
              if(getSaveKeys()) {
                  // Primary partition id
                  ByteUtils.writeInt(outputValue, partitionList.get(0), ByteUtils.SIZE_OF_INT);
  
                  // Replica type
                  ByteUtils.writeBytes(outputValue,
                                       replicaType,
                                       2 * ByteUtils.SIZE_OF_INT,
                                       ByteUtils.SIZE_OF_BYTE);
              } else {
                  // Partition id
                  ByteUtils.writeInt(outputValue,
                                     partitionList.get(replicaType),
                                     ByteUtils.SIZE_OF_INT);
              }
  
              collector.collect(outputKey, outputValue);
          }
          md5er.reset();
      }